Merge pull request #63 from amido/johnsall/sb-listener

ServiceBus Listener

build/azDevOps/azure/azure-pipelines-javaspring-k8s.yml

@@ -448,8 +448,14 @@ stages:
             value: $(kubernetes_clustername_nonprod)
           - name: app_name
             value: "java-api"
+          - name: servicebus_enabled
+            value: "true"
           - name: servicebus_connectionstring
             value: $[ dependencies.AppInfraDev.outputs['AppInfraDev.tfoutputs.servicebus_connectionstring'] ]
+          - name: servicebus_topic
+            value: $[ dependencies.AppInfraDev.outputs['AppInfraDev.tfoutputs.servicebus_topic_name'] ]
+          - name: servicebus_subscription
+            value: $[ dependencies.AppInfraDev.outputs['AppInfraDev.tfoutputs.servicebus_subscription_name'] ]
         strategy:
           runOnce:
             deploy:
@@ -491,6 +497,10 @@ stages:
                           cosmosdb_key: "$(cosmosdb_primary_master_key)",
                           cosmosdb_endpoint: "$(cosmosdb_endpoint)",
                           cosmosdb_name: "$(cosmosdb_database_name)",
+                          sb_enabled: "$(servicebus_enabled)",
+                          sb_conn_string: "$(servicebus_connectionstring)",
+                          sb_topic: "$(servicebus_topic)",
+                          sb_subscription: "$(servicebus_subscription)"
                         },
                         additional_args: "-no-empty",
                       },
@@ -501,6 +511,13 @@ stages:
                     azure_subscription_id: "$(azure-subscription-id)"
                     aks_cluster_resourcegroup: "${{ variables.aks_cluster_resourcegroup }}"
                     aks_cluster_name: "${{ variables.aks_cluster_name }}"
+                    servicebus_creds:
+                      {
+                        SB_ENABLED: true,
+                        SB_TOPIC: servicebus_topic,
+                        SB_SUBSCRIPTION: servicebus_subscription,
+                        SB_CONNECTION_STRING: servicebus_connectionstring,
+                      }
                     # Used to do a `kubectl rollout status`
                     deployments: [
                       {
@@ -531,9 +548,13 @@ stages:
                         OAUTH_TOKEN_URL: $(AUTH0_TOKEN_URL),
                         BASE_URL: "${{ variables.dns_pointer }}",
                       }
-                    service_config: {
-                      SERVICEBUS_CONNECTIONSTRING: "${{ variables.servicebus_conenctionstring }}"
-                    }
+                    servicebus_creds:
+                      {
+                        SB_ENABLED: true,
+                        SB_TOPIC: servicebus_topic,
+                        SB_SUBSCRIPTION: servicebus_subscription,
+                        SB_CONNECTION_STRING: servicebus_connectionstring,
+                      }
 
   - stage: Prod
     dependsOn: Build
@@ -714,7 +735,11 @@ stages:
           - name: app_name
             value: "java-api"
           - name: servicebus_connectionstring
-            value: $[ dependencies.AppInfraProd.outputs['AppInfraProd.tfoutputs.servicebus_connectionstring'] ]
+            value: $[ dependencies.AppInfraDev.outputs['AppInfraDev.tfoutputs.servicebus_connectionstring'] ]
+          - name: servicebus_topic
+            value: $[ dependencies.AppInfraDev.outputs['AppInfraDev.tfoutputs.servicebus_topic_name'] ]
+          - name: servicebus_subscription
+            value: $[ dependencies.AppInfraDev.outputs['AppInfraDev.tfoutputs.servicebus_subscription_name'] ]
         strategy:
           runOnce:
             deploy:
@@ -755,6 +780,10 @@ stages:
                           cosmosdb_key: "$(cosmosdb_primary_master_key)",
                           cosmosdb_endpoint: "$(cosmosdb_endpoint)",
                           cosmosdb_name: "$(cosmosdb_database_name)",
+                          sb_enabled: "$(servicebus_enabled)",
+                          sb_conn_string: "$(servicebus_connectionstring)",
+                          sb_topic: "$(servicebus_topic)",
+                          sb_subscription: "$(servicebus_subscription)"
                         },
                         additional_args: "-no-empty",
                       },
@@ -773,6 +802,13 @@ stages:
                         timeout: "120s",
                       },
                     ]
+                    servicebus_creds:
+                      {
+                        SB_ENABLED: true,
+                        SB_TOPIC: servicebus_topic,
+                        SB_SUBSCRIPTION: servicebus_subscription,
+                        SB_CONNECTION_STRING: servicebus_connectionstring,
+                      }
 
                 - template: templates/steps/deploy/deploy-post-deploy-tests.yml
                   parameters:
@@ -795,3 +831,10 @@ stages:
                         OAUTH_TOKEN_URL: $(AUTH0_TOKEN_URL),
                         BASE_URL: "${{ variables.dns_pointer }}",
                       }
+                    servicebus_creds:
+                      {
+                        SB_ENABLED: true,
+                        SB_TOPIC: servicebus_topic,
+                        SB_SUBSCRIPTION: servicebus_subscription,
+                        SB_CONNECTION_STRING: servicebus_connectionstring,
+                      }

build/azDevOps/azure/templates/steps/deploy/deploy-app.yml

@@ -10,6 +10,7 @@ parameters:
   aks_cluster_resourcegroup: ""
   aks_cluster_name: ""
   deployments: []
+  servicebus_creds: {}
 steps:
   # Template Substitutions
   - ${{ each template in parameters.template_input }}:
@@ -53,6 +54,9 @@ steps:
           arguments: >
             -a "${{ template.out }}"
         displayName: "Deploy: Kubectl Apply (${{ template.out }})"
+        env:
+          ${{ each var in parameters.servicebus_creds }}:
+            ${{ var.key }}: ${{ var.value }}
 
   - ${{ each deployment in parameters.deployments }}:
       - task: Bash@3

build/azDevOps/azure/templates/steps/deploy/deploy-post-deploy-tests-karate.yml

@@ -11,6 +11,7 @@ parameters:
   maven_post_deploy_failsafe_reports_directory: ""
   docker_java_container: ""
   auth0_credentials: {}
+  servicebus_creds: {}
 steps:
   - ${{ if eq(parameters.functional_test, true) }}:
       - download: current
@@ -45,6 +46,8 @@ steps:
         env:
           ${{ each var in parameters.auth0_credentials }}:
             ${{ var.key }}: ${{ var.value }}
+          ${{ each var in parameters.servicebus_creds }}:
+            ${{ var.key }}: ${{ var.value }}
 
       - task: PublishPipelineArtifact@1
         condition: always()

build/azDevOps/azure/templates/steps/deploy/deploy-post-deploy-tests-serenity.yml

@@ -11,6 +11,7 @@ parameters:
   maven_post_deploy_failsafe_reports_directory: ""
   docker_java_container: ""
   auth0_credentials: {}
+  servicebus_creds: {}
 steps:
   - ${{ if eq(parameters.functional_test, true) }}:
       - download: current
@@ -46,7 +47,8 @@ steps:
         env:
           ${{ each var in parameters.auth0_credentials }}:
             ${{ var.key }}: ${{ var.value }}
-
+          ${{ each var in parameters.servicebus_creds }}:
+            ${{ var.key }}: ${{ var.value }}
       # Run after all Post-Deploy tests to aggregate Serenity reports.
       - task: Bash@3
         inputs:

build/azDevOps/azure/templates/steps/deploy/deploy-post-deploy-tests.yml

@@ -11,6 +11,7 @@ parameters:
   maven_post_deploy_failsafe_reports_directory: ""
   docker_java_container: ""
   auth0_credentials: {}
+  servicebus_creds: {}
 steps:
   - template: ./deploy-post-deploy-tests-serenity.yml
     parameters:
@@ -25,6 +26,7 @@ steps:
       maven_post_deploy_html_report_directory: "${{ parameters.maven_post_deploy_html_report_directory }}"
       docker_java_container: "${{ parameters.docker_java_container }}"
       auth0_credentials: "${{ parameters.auth0_credentials }}"
+      servicebus_creds: "${{ parameters.servicebus_creds }}"
 
   - template: ./deploy-post-deploy-tests-karate.yml
     parameters:
@@ -39,3 +41,4 @@ steps:
       maven_post_deploy_html_report_directory: "target/surefire-reports"
       docker_java_container: "${{ parameters.docker_java_container }}"
       auth0_credentials: "${{ parameters.auth0_credentials }}"
+      servicebus_creds: "${{ parameters.servicebus_creds }}"

deploy/k8s/app/base_api-deploy.yml

@@ -207,6 +207,14 @@ spec:
                 secretKeyRef:
                   name: appsecrets
                   key: app_insights
+            - name: SB_ENABLED
+              value: "${sb_enabled}"
+            - name: SB_CONNECTION_STRING
+              value: "${sb_conn_string}"
+            - name: SB_SUBSCRIPTION
+              value: "${sb_subscription}"
+            - name: SB_TOPIC
+              value: "${sb_topic}"
 
           ports:
             - name: http

java/pom.xml

@@ -23,7 +23,7 @@
         <applicationinsights.version>2.6.3</applicationinsights.version>
         <cosmosdb.version>3.6.0</cosmosdb.version>
         <azure.springboot.version>3.6.0</azure.springboot.version>
-        <azure.servicebus.version>3.6.4</azure.servicebus.version>
+        <azure.messaging-servicebus.version>7.3.0</azure.messaging-servicebus.version>
         <jackson.version>2.12.3</jackson.version>
         <mockito.version>3.11.2</mockito.version>
         <lombok.version>1.18.20</lombok.version>
@@ -157,9 +157,9 @@
         </dependency>
 
         <dependency>
-            <groupId>com.microsoft.azure</groupId>
-            <artifactId>azure-servicebus</artifactId>
-            <version>${azure.servicebus.version}</version>
+            <groupId>com.azure</groupId>
+            <artifactId>azure-messaging-servicebus</artifactId>
+            <version>${azure.messaging-servicebus.version}</version>
             <exclusions>
                 <exclusion>
                     <groupId>com.nimbusds</groupId>

java/sonar-project.properties

@@ -30,4 +30,4 @@ sonar.language=java
 sonar.java.source=11
 
 sonar.cpd.exclusions=**/model/*
-sonar.coverage.exclusions=**/AuthControllerImpl*,**/ApplicationConfig*
+sonar.coverage.exclusions=**/AuthControllerImpl*,**/ApplicationConfig*,**/servicebus/*,**/messaging/**

java/src/main/java/com/amido/stacks/ApplicationConfig.java

@@ -1,6 +1,8 @@
 package com.amido.stacks;
 
 import com.auth0.spring.security.api.JwtWebSecurityConfigurer;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.Arrays;
 import org.apache.commons.lang3.BooleanUtils;
 import org.springframework.beans.factory.annotation.Value;
@@ -52,6 +54,16 @@ CorsConfigurationSource corsConfigurationSource() {
     return source;
   }
 
+  /**
+   * Provides a custom {@link ObjectMapper} that ignores extra fields.
+   *
+   * @return the modified implementation
+   */
+  @Bean
+  ObjectMapper objectMapper() {
+    return new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+  }
+
   /**
    * Configure your API to use the RS256 and protect API endpoints. Config switch is put in place to
    * enable or disable Auth (isEnabled)

java/src/main/java/com/amido/stacks/core/api/exception/ApiExceptionAdvice.java

@@ -1,10 +1,10 @@
 package com.amido.stacks.core.api.exception;
 
-import static com.google.common.collect.Maps.newHashMap;
 import static org.springframework.http.HttpStatus.BAD_REQUEST;
 import static org.springframework.web.context.request.RequestAttributes.SCOPE_REQUEST;
 
 import com.amido.stacks.core.api.dto.ErrorResponse;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -38,7 +38,7 @@ protected ResponseEntity<Object> handleMethodArgumentNotValid(
   }
 
   private Map<String, String> processFieldErrors(List<FieldError> fieldErrors) {
-    Map<String, String> errors = newHashMap();
+    Map<String, String> errors = new HashMap<>();
     for (FieldError fieldError : fieldErrors) {
       errors.put(fieldError.getField(), fieldError.getDefaultMessage());
     }

java/src/main/java/com/amido/stacks/core/azure/servicebus/ServiceBusConfiguration.java

@@ -1,11 +1,12 @@
 package com.amido.stacks.core.azure.servicebus;
 
+import com.azure.messaging.servicebus.ServiceBusClientBuilder;
+import com.azure.messaging.servicebus.ServiceBusClientBuilder.ServiceBusProcessorClientBuilder;
+import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import com.azure.messaging.servicebus.models.ServiceBusReceiveMode;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.databind.json.JsonMapper;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
-import com.microsoft.azure.servicebus.TopicClient;
-import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
-import com.microsoft.azure.servicebus.primitives.ServiceBusException;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -26,9 +27,24 @@ public ServiceBusConfiguration(ServiceBusProperties properties) {
   }
 
   @Bean
-  public TopicClient topicSender() throws ServiceBusException, InterruptedException {
-    return new TopicClient(
-        new ConnectionStringBuilder(properties.getConnectionString(), properties.getTopicName()));
+  public ServiceBusProcessorClientBuilder serviceBusProcessorClientBuilder() {
+
+    return new ServiceBusClientBuilder()
+        .connectionString(properties.getConnectionString())
+        .processor()
+        .topicName(properties.getTopicName())
+        .subscriptionName(properties.getSubscriptionName())
+        .processError(context -> context.getException().printStackTrace())
+        .receiveMode(ServiceBusReceiveMode.PEEK_LOCK);
+  }
+
+  @Bean
+  public ServiceBusSenderAsyncClient serviceBusAsyncSender() {
+    return new ServiceBusClientBuilder()
+        .connectionString(properties.getConnectionString())
+        .sender()
+        .topicName(properties.getTopicName())
+        .buildAsyncClient();
   }
 
   @Bean

java/src/main/java/com/amido/stacks/core/messaging/listen/ApplicationEventListener.java

@@ -0,0 +1,6 @@
+package com.amido.stacks.core.messaging.listen;
+
+public interface ApplicationEventListener {
+
+  public void listen();
+}

java/src/main/java/com/amido/stacks/core/messaging/listen/DefaultEventListener.java

@@ -0,0 +1,17 @@
+package com.amido.stacks.core.messaging.listen;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Component
+@ConditionalOnProperty(
+    value = "azure.servicebus.enabled",
+    havingValue = "false",
+    matchIfMissing = true)
+public class DefaultEventListener implements ApplicationEventListener {
+
+  @Override
+  public void listen() {
+    // Nothing to listen to for default listener
+  }
+}