Support IBM amqp

CHANGELOG.md

@@ -66,6 +66,11 @@ All notable changes to this project will be documented in this file.
 - New `reject_errored` output.
 - New `nats_request_reply` processor.
 - New `json_documents` scanner.
+- Field `persistent` added to the `amqp_1` output to specify whether message delivery should be persistent.
+- Add IBM amqp support
+  - Field `target_capabilities` added to the `amqp_1` output to specify sender extension capabilities.
+  - Field `message_properties_to` added to the `amqp_1` output to specify the intended destination of the message.
+  - Field `source_capabilities` added to the `amqp_1` input to specify receiver extension capabilities.
 
 ### Fixed
 

docs/modules/components/pages/inputs/amqp_1.adoc

@@ -55,6 +55,7 @@ input:
       root_cas: ""
       root_cas_file: ""
       client_certs: []
+    source_capabilities: [] # No default (optional)
     sasl:
       mechanism: none
       user: ""
@@ -326,6 +327,28 @@ password: foo
 password: ${KEY_PASSWORD}
 ```
 
+=== `source_capabilities`
+
+List of extension capabilities the receiver desires.
+
+
+*Type*: `array`
+
+
+```yml
+# Examples
+
+source_capabilities:
+  - queue
+
+source_capabilities:
+  - topic
+
+source_capabilities:
+  - queue
+  - topic
+```
+
 === `sasl`
 
 Enables SASL authentication.

docs/modules/components/pages/outputs/amqp_1.adoc

@@ -63,6 +63,9 @@ output:
       password: ""
     metadata:
       exclude_prefixes: []
+    persistent: false
+    target_capabilities: [] # No default (optional)
+    message_properties_to: amqp://localhost:5672/ # No default (optional)
 ```
 
 --
@@ -377,4 +380,49 @@ Provide a list of explicit metadata key prefixes to be excluded when adding meta
 
 *Default*: `[]`
 
+=== `persistent`
+
+Whether message delivery should be persistent (transient by default).
+
+
+*Type*: `bool`
+
+*Default*: `false`
+
+=== `target_capabilities`
+
+List of extension capabilities the sender desires.
+
+
+*Type*: `array`
+
+
+```yml
+# Examples
+
+target_capabilities:
+  - queue
+
+target_capabilities:
+  - topic
+
+target_capabilities:
+  - queue
+  - topic
+```
+
+=== `message_properties_to`
+
+Identifies the node that is the intended destination of the message.
+
+
+*Type*: `string`
+
+
+```yml
+# Examples
+
+message_properties_to: amqp://localhost:5672/
+```
+
 

internal/impl/amqp1/config.go

@@ -37,11 +37,15 @@ const (
 	azureRenewLockField   = "azure_renew_lock"
 	getMessageHeaderField = "read_header"
 	creditField           = "credit"
+	sourceCapsField       = "source_capabilities"
 
 	// Output
 	targetAddrField  = "target_address"
 	appPropsMapField = "application_properties_map"
 	metaFilterField  = "metadata"
+	persistentField  = "persistent"
+	targetCapsField  = "target_capabilities"
+	messagePropsTo   = "message_properties_to"
 )
 
 // ErrSASLMechanismNotSupported is returned if a SASL mechanism was not recognised.

internal/impl/amqp1/input.go

@@ -75,6 +75,13 @@ func amqp1InputSpec() *service.ConfigSpec {
 				Default(64).
 				Advanced(),
 			service.NewTLSToggledField(tlsField),
+			service.NewStringListField(sourceCapsField).
+				Description("List of extension capabilities the receiver desires.").
+				Optional().
+				Advanced().
+				Example([]string{"queue"}).
+				Example([]string{"topic"}).
+				Example([]string{"queue", "topic"}),
 			saslFieldSpec(),
 		).LintRule(`
 root = if this.url.or("") == "" && this.urls.or([]).length() == 0 {
@@ -101,7 +108,9 @@ type amqp1Reader struct {
 	renewLock  bool
 	getHeader  bool
 	credit     int // max_in_flight
+	srcCaps    []string
 	connOpts   *amqp.ConnOptions
+	recvOpts   *amqp.ReceiverOptions
 	log        *service.Logger
 
 	m    sync.RWMutex
@@ -112,6 +121,7 @@ func amqp1ReaderFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 	a := amqp1Reader{
 		log:      mgr.Logger(),
 		connOpts: &amqp.ConnOptions{},
+		recvOpts: &amqp.ReceiverOptions{},
 	}
 
 	urlStrs, err := conf.FieldStringList(urlsField)
@@ -135,7 +145,6 @@ func amqp1ReaderFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 		}
 
 		a.urls = append(a.urls, singleURL)
-
 	}
 
 	if a.sourceAddr, err = conf.FieldString(sourceAddrField); err != nil {
@@ -153,6 +162,7 @@ func amqp1ReaderFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 	if a.credit, err = conf.FieldInt(creditField); err != nil {
 		return nil, err
 	}
+	a.recvOpts.Credit = int32(a.credit)
 
 	if err := saslOptFnsFromParsed(conf, a.connOpts); err != nil {
 		return nil, err
@@ -166,6 +176,14 @@ func amqp1ReaderFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 		a.connOpts.TLSConfig = tlsConf
 	}
 
+	a.srcCaps, err = conf.FieldStringList(sourceCapsField)
+	if err != nil {
+		return nil, err
+	}
+	if len(a.srcCaps) != 0 {
+		a.recvOpts.SourceCapabilities = a.srcCaps
+	}
+
 	return &a, nil
 }
 
@@ -194,9 +212,7 @@ func (a *amqp1Reader) Connect(ctx context.Context) (err error) {
 	}
 
 	// Create a receiver
-	if conn.receiver, err = conn.session.NewReceiver(ctx, a.sourceAddr, &amqp.ReceiverOptions{
-		Credit: int32(a.credit),
-	}); err != nil {
+	if conn.receiver, err = conn.session.NewReceiver(ctx, a.sourceAddr, a.recvOpts); err != nil {
 		_ = conn.Close(ctx)
 		return
 	}

internal/impl/amqp1/output.go

@@ -68,6 +68,22 @@ This output benefits from sending multiple messages in flight in parallel for im
 			saslFieldSpec(),
 			service.NewMetadataExcludeFilterField(metaFilterField).
 				Description("Specify criteria for which metadata values are attached to messages as headers."),
+			service.NewBoolField(persistentField).
+				Description("Whether message delivery should be persistent (transient by default).").
+				Advanced().
+				Default(false),
+			service.NewStringListField(targetCapsField).
+				Description("List of extension capabilities the sender desires.").
+				Optional().
+				Advanced().
+				Example([]string{"queue"}).
+				Example([]string{"topic"}).
+				Example([]string{"queue", "topic"}),
+			service.NewStringField(messagePropsTo).
+				Description("Identifies the node that is the intended destination of the message.").
+				Optional().
+				Advanced().
+				Example("amqp://localhost:5672/"),
 		).LintRule(`
 root = if this.url.or("") == "" && this.urls.or([]).length() == 0 {
   "field 'urls' must be set"
@@ -105,15 +121,20 @@ type amqp1Writer struct {
 	metaFilter               *service.MetadataExcludeFilter
 	applicationPropertiesMap *bloblang.Executor
 	connOpts                 *amqp.ConnOptions
+	senderOpts               *amqp.SenderOptions
+	persistent               bool
+	targetCaps               []string
+	msgTo                    string
 
 	log      *service.Logger
 	connLock sync.RWMutex
 }
 
 func amqp1WriterFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (*amqp1Writer, error) {
 	a := amqp1Writer{
-		log:      mgr.Logger(),
-		connOpts: &amqp.ConnOptions{},
+		log:        mgr.Logger(),
+		connOpts:   &amqp.ConnOptions{},
+		senderOpts: &amqp.SenderOptions{},
 	}
 
 	urlStrs, err := conf.FieldStringList(urlsField)
@@ -164,6 +185,25 @@ func amqp1WriterFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 	if a.metaFilter, err = conf.FieldMetadataExcludeFilter(metaFilterField); err != nil {
 		return nil, err
 	}
+
+	if a.persistent, err = conf.FieldBool(persistentField); err != nil {
+		return nil, err
+	}
+
+	a.targetCaps, err = conf.FieldStringList(targetCapsField)
+	if err != nil {
+		return nil, err
+	}
+	if len(a.targetCaps) != 0 {
+		a.senderOpts.TargetCapabilities = a.targetCaps
+	}
+
+	if conf.Contains(messagePropsTo) {
+		if a.msgTo, err = conf.FieldString(messagePropsTo); err != nil {
+			return nil, err
+		}
+	}
+
 	return &a, nil
 }
 
@@ -193,7 +233,7 @@ func (a *amqp1Writer) Connect(ctx context.Context) (err error) {
 	}
 
 	// Create a sender
-	if sender, err = session.NewSender(ctx, a.targetAddr, nil); err != nil {
+	if sender, err = session.NewSender(ctx, a.targetAddr, a.senderOpts); err != nil {
 		_ = session.Close(ctx)
 		_ = client.Close()
 		return
@@ -280,6 +320,14 @@ func (a *amqp1Writer) Write(ctx context.Context, msg *service.Message) error {
 		return nil
 	})
 
+	if a.persistent {
+		m.Header = &amqp.MessageHeader{Durable: true}
+	}
+
+	if a.msgTo != "" {
+		m.Properties = &amqp.MessageProperties{To: &a.msgTo}
+	}
+
 	if err = s.Send(ctx, m, nil); err != nil {
 		if ctx.Err() == nil {
 			a.log.Errorf("Lost connection due to: %v\n", err)