Add service bus partition key sample. (#24)

Author: ZhuXiaoBing-cn

servicebus/azure-spring-cloud-stream-binder-servicebus-queue/servicebus-queue-binder/README.md

@@ -164,23 +164,110 @@ spring:
 
 
 ## Examples
+### How to configure the partition key
+**Example: Manually set the partition key of the message through application.yml**
 
-1.  Run the `mvn spring-boot:run` in the root of the code sample to get the app running.
+This example demonstrates how to manually set the partition key for the message in the application.
 
-1.  Send a POST request
+**Way 1:**
+This example requires that `spring.cloud.stream.default.producer.partitionKeyExpression` be set `"'partitionKey-' + headers[<message-header-key>]"`.
+```yaml
+spring:
+  cloud:
+    azure:
+      servicebus:
+        connection-string: [servicebus-namespace-connection-string]
+    stream:
+      default:
+        producer:
+          partitionKeyExpression:  "'partitionKey-' + headers[<message-header-key>]"
+```
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader("<message-header-key>", "Customize partirion key")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+
+> **NOTE:** When using `application.yml` to configure the partition key, its priority will be the lowest.
+> It will take effect only when the `ServiceBusMessageHeaders.SESSION_ID`, `ServiceBusMessageHeaders.PARTITION_KEY`, `AzureHeaders.PARTITION_KEY` are not configured.
+
+**Way 2:**
+Manually add the partition Key in the message header by code.
+
+*Recommended:* Use `ServiceBusMessageHeaders.PARTITION_KEY` as the key of the header.
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partirion key")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+
+*Not recommended but currently supported:* `AzureHeaders.PARTITION_KEY` as the key of the header.
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader(AzureHeaders.PARTITION_KEY, "Customize partirion key")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+> **NOTE:** When both `ServiceBusMessageHeaders.PARTITION_KEY` and `AzureHeaders.PARTITION_KEY` are set in the message headers,
+> `ServiceBusMessageHeaders.PARTITION_KEY` is preferred.
+
+### How to configure session id
+**Example: Set the session id for the message**
+
+This example demonstrates how to manually set the session id of a message in the application.
+
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session id")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+
+> **NOTE:** When the `ServiceBusMessageHeaders.SESSION_ID` is set in the message headers, and a different `ServiceBusMessageHeaders.PARTITION_KEY` (or `AzureHeaders.PARTITION_KEY`) header is also set,
+> the value of the session id will eventually be used to overwrite the value of the partition key.
+
+### Run sample
+1. Run the `mvn spring-boot:run` in the root of the code sample to get the app running.
+
+2. Send a POST request
 
         $ curl -X POST http://localhost:8080/messages?message=hello
 
     or when the app runs on App Service or VM
 
         $ curl -d -X POST https://[your-app-URL]/messages?message=hello
 
-1.  Verify in your app’s logs that a similar message was posted:
+3. Verify in your app’s logs that a similar message was posted:
 
         New message received: 'hello'
         Message 'hello' successfully checkpointed
 
-1.  Delete the resources on [Azure Portal][azure-portal] to avoid unexpected charges.
+4. Send a POST request
+
+        $ curl -X POST http://localhost:8080/setSessionId?message=hello
+
+5. It will be visible on Azure Portal that sent messages are configured with session id and partition key.
+   ![Azure portal displays the session id.](docs/image-set-session-id.png )
+
+6. Delete the resources on [Azure Portal][azure-portal] to avoid unexpected charges.
 
 ## Enhancement
 ### Set Service Bus message headers

servicebus/azure-spring-cloud-stream-binder-servicebus-queue/servicebus-queue-binder/docs/image-set-session-id.png

@@ -3,6 +3,7 @@
 
 package com.azure.spring.sample.servicebus.queue.binder;
 
+import com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,6 +33,30 @@ public ResponseEntity<String> sendMessage(@RequestParam String message) {
         return ResponseEntity.ok("Sent!");
     }
 
+    /**
+     * Set the session id scene.
+     */
+    @PostMapping("/setSessionId")
+    public ResponseEntity<String> setSessionId(@RequestParam String message) {
+        LOGGER.info("Going to add message {} to Sinks.Many.", message);
+        many.emitNext(MessageBuilder.withPayload(message)
+                                    .setHeader(ServiceBusMessageHeaders.SESSION_ID, "<custom-session-id>")
+                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+        return ResponseEntity.ok("Sent!");
+    }
+
+    /**
+     * Set the ServiceBusMessageHeaders partition key scene.
+     */
+    @PostMapping("/setServiceBusMessageHeadersPartitionKey")
+    public ResponseEntity<String> setServiceBusMessageHeadersPartitionKey(@RequestParam String message) {
+        LOGGER.info("Going to add message {} to Sinks.Many.", message);
+        many.emitNext(MessageBuilder.withPayload(message)
+                                    .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "<custom-partition-key>")
+                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+        return ResponseEntity.ok("Sent!");
+    }
+
     @GetMapping("/")
     public String welcome() {
         return "welcome";

servicebus/azure-spring-cloud-stream-binder-servicebus-queue/servicebus-queue-binder/src/main/java/com/azure/spring/sample/servicebus/queue/binder/ServiceProducerController.java

@@ -165,6 +165,87 @@ spring:
 
 
 ## Examples
+### How to configure the partition key
+**Example: Manually set the partition key of the message through application.yml**
+
+This example demonstrates how to manually set the partition key for the message in the application.
+
+**Way 1:**
+This example requires that `spring.cloud.stream.default.producer.partitionKeyExpression` be set `"'partitionKey-' + headers[<message-header-key>]"`.
+```yaml
+spring:
+  cloud:
+    azure:
+      servicebus:
+        connection-string: [servicebus-namespace-connection-string]
+    stream:
+      default:
+        producer:
+          partitionKeyExpression:  "'partitionKey-' + headers[<message-header-key>]"
+```
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader("<message-header-key>", "Customize partirion key")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+
+> **NOTE:** When using `application.yml` to configure the partition key, its priority will be the lowest.
+> It will take effect only when the `ServiceBusMessageHeaders.SESSION_ID`, `ServiceBusMessageHeaders.PARTITION_KEY`, `AzureHeaders.PARTITION_KEY` are not configured.
+
+**Way 2:**
+Manually add the partition Key in the message header by code.
+
+*Recommended:* Use `ServiceBusMessageHeaders.PARTITION_KEY` as the key of the header.
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partirion key")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+
+*Not recommended but currently supported:* `AzureHeaders.PARTITION_KEY` as the key of the header.
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader(AzureHeaders.PARTITION_KEY, "Customize partirion key")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+> **NOTE:** When both `ServiceBusMessageHeaders.PARTITION_KEY` and `AzureHeaders.PARTITION_KEY` are set in the message headers,
+> `ServiceBusMessageHeaders.PARTITION_KEY` is preferred.
+
+### How to configure session id
+**Example: Set the session id for the message**
+
+This example demonstrates how to manually set the session id of a message in the application.
+
+```java
+@PostMapping("/messages")
+public ResponseEntity<String> sendMessage(@RequestParam String message) {
+    LOGGER.info("Going to add message {} to Sinks.Many.", message);
+    many.emitNext(MessageBuilder.withPayload(message)
+                                .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session id")
+                                .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+    return ResponseEntity.ok("Sent!");
+}
+```
+
+> **NOTE:** When the `ServiceBusMessageHeaders.SESSION_ID` is set in the message headers, and a different `ServiceBusMessageHeaders.PARTITION_KEY` (or `AzureHeaders.PARTITION_KEY`) header is also set,
+> the value of the session id will eventually be used to overwrite the value of the partition key.
+
+### Run sample
 
 1.  Run the `mvn spring-boot:run` in the root of the code sample to get the app running.
 
@@ -181,7 +262,14 @@ spring:
         New message received: 'hello'
         Message 'hello' successfully checkpointed
 
-1.  Delete the resources on [Azure Portal][azure-portal] to avoid unexpected charges.
+4. Send a POST request
+
+        $ curl -X POST http://localhost:8080/setSessionId?message=hello
+
+5. It will be visible on Azure Portal that sent messages are configured with session id and partition key.
+   ![Azure portal displays the session id.](docs/image-set-session-id.png )
+
+6. Delete the resources on [Azure Portal][azure-portal] to avoid unexpected charges.
 
 ## Enhancement
 ### Set Service Bus message headers

servicebus/azure-spring-cloud-stream-binder-servicebus-topic/servicebus-topic-binder/README.md

@@ -3,6 +3,7 @@
 
 package com.azure.spring.sample.servicebus.topic.binder;
 
+import com.azure.spring.integration.servicebus.converter.ServiceBusMessageHeaders;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,6 +33,30 @@ public ResponseEntity<String> sendMessage(@RequestParam String message) {
         return ResponseEntity.ok("Sent!");
     }
 
+    /**
+     * Set the session id scene.
+     */
+    @PostMapping("/setSessionId")
+    public ResponseEntity<String> setSessionId(@RequestParam String message) {
+        LOGGER.info("Going to add message {} to Sinks.Many.", message);
+        many.emitNext(MessageBuilder.withPayload(message)
+                                    .setHeader(ServiceBusMessageHeaders.SESSION_ID, "<custom-session-id>")
+                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+        return ResponseEntity.ok("Sent!");
+    }
+
+    /**
+     * Set the ServiceBusMessageHeaders partition key scene.
+     */
+    @PostMapping("/setServiceBusMessageHeadersPartitionKey")
+    public ResponseEntity<String> setServiceBusMessageHeadersPartitionKey(@RequestParam String message) {
+        LOGGER.info("Going to add message {} to Sinks.Many.", message);
+        many.emitNext(MessageBuilder.withPayload(message)
+                                    .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "<custom-partition-key>")
+                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
+        return ResponseEntity.ok("Sent!");
+    }
+
     @GetMapping("/")
     public String welcome() {
         return "welcome";