Merge Code and Updated Release Version

.github/workflows/docker-build-push-dev-master.yml

@@ -0,0 +1,32 @@
+name: Docker Build
+on:
+  push:
+    branches:
+      - master
+jobs:
+  docker-build-push:
+    runs-on: ubuntu-20.04
+    timeout-minutes: 40
+    steps:
+      - name: Check out the repo
+        uses: actions/checkout@v3
+      - name: Extract version from pom.xml
+        id: version
+        run: |
+          echo ::set-output name=VERSION::$(grep -A 1 '<artifactId>orchestrator</artifactId>' pom.xml | grep -oP '(?<=<version>).*?(?=</version>)')
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v2
+      - name: Log in to Docker Hub
+        uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
+        with:
+          username: ${{secrets.DOCKER_HUB_USERNAME}}
+          password: ${{secrets.DOCKER_HUB_ACCESS_TOKEN}}
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v2
+        with:
+          context: "."
+          push: true
+          tags: samagragovernance/orchestrator:v${{ steps.version.outputs.VERSION }}
+          build-args: |
+            username=${{ secrets.USERNAME }},
+            token=${{ secrets.TOKEN }}

.github/workflows/docker-build-push-dev.yml

@@ -0,0 +1,32 @@
+name: Docker Build
+on:
+  push:
+    branches:
+      - development
+jobs:
+  docker-build-push:
+    runs-on: ubuntu-20.04
+    timeout-minutes: 40
+    steps:
+      - name: Check out the repo
+        uses: actions/checkout@v3
+      - name: Extract version from pom.xml
+        id: version
+        run: |
+          echo ::set-output name=VERSION::$(grep -A 1 '<artifactId>orchestrator</artifactId>' pom.xml | grep -oP '(?<=<version>).*?(?=</version>)')
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v2
+      - name: Log in to Docker Hub
+        uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
+        with:
+          username: ${{secrets.DOCKER_HUB_USERNAME}}
+          password: ${{secrets.DOCKER_HUB_ACCESS_TOKEN}}
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v2
+        with:
+          context: "."
+          push: true
+          tags: samagragovernance/orchestrator:v${{ steps.version.outputs.VERSION }}-SNAPSHOT
+          build-args: |
+            username=${{ secrets.USERNAME }},
+            token=${{ secrets.TOKEN }}

Dockerfile

@@ -28,11 +28,11 @@ ADD /src $HOME/src
 RUN mvn package -s $HOME/settings.xml -DskipTests=true
 
 # Package stage
-FROM openjdk:11
+FROM ibm-semeru-runtimes:open-11.0.18_10-jre
 ENV HOME=/home/app
 ENV export $(cat .env | xargs)
 WORKDIR $HOME
 COPY --from=build $HOME/target/*.jar app.jar
 
 EXPOSE 8080
-ENTRYPOINT ["java","-jar","app.jar"]
+ENTRYPOINT ["java","-Xmx1024m","-Xshareclasses","-XX:+CMSClassUnloadingEnabled","-XX:+UseG1GC","-XX:+ExplicitGCInvokesConcurrent","-jar","app.jar"]

pom.xml

@@ -11,15 +11,15 @@
 	</parent>
 	<groupId>com.uci</groupId>
 	<artifactId>orchestrator</artifactId>
-	<version>2.2.1</version>
+	<version>2.2.9</version>
 	<name>orchestrator</name>
 	<description>Demo project for Spring Boot</description>
 
 	<properties>
 		<java.version>11</java.version>
-		<utils.version>2.2.1</utils.version>
-		<messagerosa.version>2.2.1</messagerosa.version>
-		<dao.version>2.2.1</dao.version>
+		<utils.version>2.2.4</utils.version>
+		<messagerosa.version>2.2.4</messagerosa.version>
+		<dao.version>2.2.4</dao.version>
 	</properties>
 
 	<!-- For Downloading repositories from github packages -->

src/main/java/com/uci/orchestrator/Application/AppConfigOrchestrator.java

@@ -3,11 +3,13 @@
 import com.github.benmanes.caffeine.cache.Cache;
 import com.uci.orchestrator.Drools.DroolsBeanFactory;
 import com.uci.utils.BotService;
+import com.uci.utils.dto.BotServiceParams;
 import com.uci.utils.kafka.ReactiveProducer;
 import io.fusionauth.client.FusionAuthClient;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.kie.api.io.Resource;
 import org.kie.api.runtime.KieSession;
 import org.kie.internal.io.ResourceFactory;
@@ -16,9 +18,7 @@
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.kafka.core.DefaultKafkaProducerFactory;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.core.*;
 import org.springframework.kafka.support.serializer.JsonDeserializer;
 import org.springframework.web.reactive.function.client.WebClient;
 import reactor.core.publisher.Flux;
@@ -43,9 +43,9 @@ public class AppConfigOrchestrator {
 
     @Value("${campaign.url}")
     public String CAMPAIGN_URL;
-    
+
     @Value("${campaign.admin.token}")
-	public String CAMPAIGN_ADMIN_TOKEN;
+    public String CAMPAIGN_ADMIN_TOKEN;
 
     @Value("${fusionauth.url}")
     public String FUSIONAUTH_URL;
@@ -88,6 +88,8 @@ Map<String, Object> kafkaConsumerConfiguration() {
         configuration.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.springframework.kafka.support.serializer.JsonSerializer.class);
         configuration.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.springframework.kafka.support.serializer.JsonSerializer.class);
         configuration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+        configuration.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+        configuration.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "1800000");
         return configuration;
     }
 
@@ -120,7 +122,7 @@ SenderOptions<Integer, String> kafkaSenderOptions() {
 
     @Bean
     Flux<ReceiverRecord<String, String>> reactiveKafkaReceiver(ReceiverOptions<String, String> kafkaReceiverOptions) {
-        return KafkaReceiver.create(kafkaReceiverOptions).receive();
+        return KafkaReceiver.create(kafkaReceiverOptions).receive().doOnNext(r -> r.receiverOffset().acknowledge());
     }
 
     @Bean
@@ -132,21 +134,22 @@ KafkaSender<Integer, String> reactiveKafkaSender(SenderOptions<Integer, String>
     ReactiveProducer kafkaReactiveProducer() {
         return new ReactiveProducer();
     }
-    
+
     @Bean
-    ProducerFactory<String, String> producerFactory(){
-    	ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(kafkaProducerConfiguration());
-    	return producerFactory;
+    ProducerFactory<String, String> producerFactory() {
+        ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(kafkaProducerConfiguration());
+        return producerFactory;
     }
-    
+
     @Bean
     KafkaTemplate<String, String> kafkaTemplate() {
-    	KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
-    	return (KafkaTemplate<String, String>) kafkaTemplate;
+        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
+        return (KafkaTemplate<String, String>) kafkaTemplate;
     }
 
     /**
      * Create process outbound topic, if does not exists
+     *
      * @return
      */
     @Bean
@@ -156,6 +159,7 @@ public NewTopic createProcessOutboundTopic() {
 
     /**
      * Create broadcast transformer topic, if does not exists
+     *
      * @return
      */
     @Bean
@@ -165,10 +169,16 @@ public NewTopic createBroadcastTransformerTopic() {
 
     /**
      * Create generic transformer topic, if does not exists
+     *
      * @return
      */
     @Bean
     public NewTopic createGenericTransformerTopic() {
         return new NewTopic(genericTransformerTopic, 1, (short) 1);
     }
+
+    @Bean
+    public BotServiceParams getBotServiceParams() {
+        return new BotServiceParams();
+    }
 }

src/main/java/com/uci/orchestrator/Consumer/OuboundConsumer.java

@@ -22,15 +22,33 @@ public class OuboundConsumer {
     @Value("${outbound}")
     private String outboundTopic;
 
+    @Value("${notificationOutbound}")
+    public String notificationOutbound;
+
     @Autowired
     public SimpleProducer kafkaProducer;
 
+    private long consumeCount, pushNotificationCount, pushOtherCount;
+
+
     @KafkaListener(id = "${processOutbound}", topics = "${processOutbound}", properties = {"spring.json.value.default.type=java.lang.String"})
     public void onMessage(@Payload String stringMessage) {
         try {
+            consumeCount++;
+            log.info("OutboundConsumer:onMessage:: Consume Topic Count: " + consumeCount);
             XMessage msg = XMessageParser.parse(new ByteArrayInputStream(stringMessage.getBytes()));
-            kafkaProducer.send(outboundTopic, msg.toXML());
-        } catch(JAXBException e) {
+            if (msg != null && msg.getProvider() != null && msg.getProvider().equalsIgnoreCase("firebase")
+                    && msg.getChannel().equalsIgnoreCase("web")) {
+                pushNotificationCount++;
+                log.info("OutboundConsumer:onMessage:: Notification push to kafka topic count: " + pushNotificationCount + " UserId : " + msg.getTo().getUserID());
+                kafkaProducer.send(notificationOutbound, msg.toXML());
+            } else {
+                pushOtherCount++;
+                log.info("OutboundConsumer:onMessage:: Other push to kafka topic count: " + pushOtherCount);
+                kafkaProducer.send(outboundTopic, msg.toXML());
+            }
+
+        } catch (JAXBException e) {
             e.printStackTrace();
         }
     }