#3095 AMQP 1.0 Output - Support for message body type

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

@@ -71,6 +71,7 @@ output:
       password: ""
     metadata:
       exclude_prefixes: []
+    content_type: opaque_binary
 ```
 
 --
@@ -385,4 +386,20 @@ Provide a list of explicit metadata key prefixes to be excluded when adding meta
 
 *Default*: `[]`
 
+=== `content_type`
+
+Define the message body content type.
+
+The option `string` will transfer the message as an AMQP value of type string. Consider choosing the option `string` if your intention is to transfer UTF-8 string messages (like JSON messages) to the destination.
+
+
+*Type*: `string`
+
+*Default*: `"opaque_binary"`
+
+Options:
+`opaque_binary`
+, `string`
+.
+
 

internal/impl/amqp1/config.go

@@ -42,6 +42,15 @@ const (
 	targetAddrField  = "target_address"
 	appPropsMapField = "application_properties_map"
 	metaFilterField  = "metadata"
+	contentTypeField = "content_type"
+)
+
+// Content Type Options
+const (
+	// Data section with opaque binary data
+	amqpContentTypeOpaqueBinary = "opaque_binary"
+	// Single AMQP string value
+	amqpContentTypeString = "string"
 )
 
 // ErrSASLMechanismNotSupported is returned if a SASL mechanism was not recognised.

internal/impl/amqp1/integration_test.go

@@ -90,6 +90,21 @@ input:
     source_address: "queue:/$ID"
 `
 
+	templateWithContentTypeString := `
+output:
+  amqp_1:
+    url: amqp://guest:guest@localhost:$PORT/
+    target_address: "queue:/$ID"
+    max_in_flight: $MAX_IN_FLIGHT
+    content_type: "string"
+    metadata:
+      exclude_prefixes: [ $OUTPUT_META_EXCLUDE_PREFIX ]
+input:
+  amqp_1:
+    url: amqp://guest:guest@localhost:$PORT/
+    source_address: "queue:/$ID"
+`
+
 	testcases := []struct {
 		label    string
 		template string
@@ -102,6 +117,10 @@ input:
 			label:    "should handle new field urls",
 			template: templateWithFieldURLS,
 		},
+		{
+			label:    "should handle content type string",
+			template: templateWithContentTypeString,
+		},
 	}
 
 	for _, tc := range testcases {

internal/impl/amqp1/output.go

@@ -68,6 +68,11 @@ 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.NewStringEnumField(contentTypeField,
+				amqpContentTypeOpaqueBinary, amqpContentTypeString).
+				Description("Define the message body content type.\n\nThe option `string` will transfer the message as an AMQP value of type string. Consider choosing the option `string` if your intention is to transfer UTF-8 string messages (like JSON messages) to the destination.").
+				Advanced().
+				Default(amqpContentTypeOpaqueBinary),
 		).LintRule(`
 root = if this.url.or("") == "" && this.urls.or([]).length() == 0 {
   "field 'urls' must be set"
@@ -105,6 +110,7 @@ type amqp1Writer struct {
 	metaFilter               *service.MetadataExcludeFilter
 	applicationPropertiesMap *bloblang.Executor
 	connOpts                 *amqp.ConnOptions
+	contentType              string
 
 	log      *service.Logger
 	connLock sync.RWMutex
@@ -164,6 +170,11 @@ func amqp1WriterFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 	if a.metaFilter, err = conf.FieldMetadataExcludeFilter(metaFilterField); err != nil {
 		return nil, err
 	}
+
+	if a.contentType, err = conf.FieldString(contentTypeField); err != nil {
+		return nil, err
+	}
+
 	return &a, nil
 }
 
@@ -248,7 +259,16 @@ func (a *amqp1Writer) Write(ctx context.Context, msg *service.Message) error {
 		return err
 	}
 
-	m := amqp.NewMessage(mBytes)
+	m := &amqp.Message{}
+
+	switch a.contentType {
+	case amqpContentTypeOpaqueBinary:
+		m.Data = [][]byte{mBytes}
+	case amqpContentTypeString:
+		m.Value = string(mBytes)
+	default:
+		return fmt.Errorf("invalid content type specified: %s", a.contentType)
+	}
 
 	if a.applicationPropertiesMap != nil {
 		mapMsg, err := msg.BloblangQuery(a.applicationPropertiesMap)