Decouple AAE from spring-cloud-stream

activiti-cloud-audit-service/activiti-cloud-services-audit-api/src/main/java/org/activiti/cloud/services/audit/api/config/AuditAPIAutoConfiguration.java

@@ -18,8 +18,10 @@
 import org.activiti.cloud.services.audit.api.resources.EventsLinkRelationProvider;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 
 @Configuration
+@Import(AuditConsumerChannelsConfiguration.class)
 public class AuditAPIAutoConfiguration {
 
     @Bean

activiti-cloud-audit-service/activiti-cloud-services-audit-api/src/main/java/org/activiti/cloud/services/audit/api/config/AuditConsumerChannelsConfiguration.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2017-2020 Alfresco Software, Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.activiti.cloud.services.audit.api.config;
+
+import org.activiti.cloud.services.audit.api.streams.AuditConsumerChannels;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.dsl.MessageChannels;
+import org.springframework.messaging.SubscribableChannel;
+
+@Configuration
+public class AuditConsumerChannelsConfiguration implements AuditConsumerChannels {
+
+    @Bean(AUDIT_CONSUMER)
+    @Override
+    public SubscribableChannel auditConsumer() {
+        return MessageChannels.publishSubscribe(AUDIT_CONSUMER)
+            .get();
+    }
+}

activiti-cloud-audit-service/activiti-cloud-services-audit-api/src/main/java/org/activiti/cloud/services/audit/api/streams/AuditConsumerChannelHandler.java

@@ -16,13 +16,11 @@
 package org.activiti.cloud.services.audit.api.streams;
 
 import org.activiti.cloud.api.model.shared.events.CloudRuntimeEvent;
-import org.springframework.cloud.stream.annotation.StreamListener;
 import org.springframework.messaging.handler.annotation.Headers;
 
 import java.util.Map;
 
 public interface AuditConsumerChannelHandler {
 
-    @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER)
     void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events);
 }

activiti-cloud-audit-service/activiti-cloud-services-audit-api/src/main/java/org/activiti/cloud/services/audit/api/streams/AuditConsumerChannels.java

@@ -22,6 +22,5 @@ public interface AuditConsumerChannels {
 
     String AUDIT_CONSUMER = "auditConsumer";
 
-    @Input(AUDIT_CONSUMER)
     SubscribableChannel auditConsumer();
 }

activiti-cloud-audit-service/activiti-cloud-services-audit-jpa/pom.xml

@@ -18,6 +18,10 @@
       <groupId>org.activiti.cloud</groupId>
       <artifactId>activiti-cloud-service-common-config</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.activiti.cloud</groupId>
+      <artifactId>activiti-cloud-service-messaging-config</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.activiti.cloud</groupId>
       <artifactId>activiti-cloud-services-audit-api</artifactId>

activiti-cloud-audit-service/activiti-cloud-services-audit-jpa/src/main/java/org/activiti/cloud/services/audit/jpa/streams/AuditConsumerChannelHandlerImpl.java

@@ -25,7 +25,6 @@
 import org.activiti.cloud.services.audit.jpa.repository.EventsRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.cloud.stream.annotation.StreamListener;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.handler.annotation.Headers;
 import org.springframework.transaction.annotation.Propagation;
@@ -54,7 +53,6 @@ public AuditConsumerChannelHandlerImpl(EventsRepository eventsRepository,
 
     @SuppressWarnings("unchecked")
     @Override
-    @StreamListener(AuditConsumerChannels.AUDIT_CONSUMER)
     public void receiveCloudRuntimeEvent(@Headers Map<String, Object> headers, CloudRuntimeEvent<?, ?>... events) {
         if (events != null) {
             AtomicInteger counter = new AtomicInteger(0);

activiti-cloud-audit-service/activiti-cloud-services-audit-jpa/src/main/java/org/activiti/cloud/services/audit/jpa/streams/config/AuditJPAStreamsAutoConfiguration.java

@@ -15,18 +15,23 @@
  */
 package org.activiti.cloud.services.audit.jpa.streams.config;
 
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Consumer;
+import org.activiti.cloud.api.model.shared.events.CloudRuntimeEvent;
+import org.activiti.cloud.common.messaging.functional.FunctionBinding;
 import org.activiti.cloud.services.audit.api.converters.APIEventToEntityConverters;
 import org.activiti.cloud.services.audit.api.streams.AuditConsumerChannelHandler;
 import org.activiti.cloud.services.audit.api.streams.AuditConsumerChannels;
 import org.activiti.cloud.services.audit.jpa.repository.EventsRepository;
 import org.activiti.cloud.services.audit.jpa.streams.AuditConsumerChannelHandlerImpl;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
 
 @Configuration
-@EnableBinding(AuditConsumerChannels.class)
 public class AuditJPAStreamsAutoConfiguration {
 
     @Bean
@@ -37,4 +42,15 @@ public AuditConsumerChannelHandler auditConsumerChannelHandler(EventsRepository
                                                    eventConverters);
     }
 
+    @FunctionBinding(input = AuditConsumerChannels.AUDIT_CONSUMER)
+    @Bean
+    public Consumer<Message<List<CloudRuntimeEvent<?, ?>>>> auditConsumerChannelHandlerConsumer(AuditConsumerChannelHandler handler) {
+        return message -> {
+            handler.receiveCloudRuntimeEvent(message.getHeaders(),
+                Optional.ofNullable(message.getPayload())
+                    .orElse(Collections.emptyList())
+                    .toArray(new CloudRuntimeEvent[0]));
+        };
+    }
+
 }

activiti-cloud-connectors/activiti-cloud-starter-connector/pom.xml

@@ -80,8 +80,10 @@
     </dependency>
     <dependency>
       <groupId>org.springframework.cloud</groupId>
-      <artifactId>spring-cloud-stream-binder-rabbit-test-support</artifactId>
+      <artifactId>spring-cloud-stream</artifactId>
       <scope>test</scope>
+      <classifier>test-binder</classifier>
+      <type>test-jar</type>
     </dependency>
     <dependency>
       <groupId>org.springframework.boot</groupId>

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/channels/IntegrationErrorChannelResolver.java

@@ -20,6 +20,6 @@
 
 public interface IntegrationErrorChannelResolver {
 
-    MessageChannel resolveDestination(IntegrationRequest event);
+    String resolveDestination(IntegrationRequest event);
 
 }

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/channels/IntegrationErrorChannelResolverImpl.java

@@ -16,26 +16,20 @@
 package org.activiti.cloud.connectors.starter.channels;
 
 import org.activiti.cloud.api.process.model.IntegrationRequest;
-import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
-import org.springframework.messaging.MessageChannel;
 
 public class IntegrationErrorChannelResolverImpl implements IntegrationErrorChannelResolver {
 
-    private final BinderAwareChannelResolver resolver;
-
     private final IntegrationErrorDestinationBuilder integrationErrorDestinationBuilder;
 
-    public IntegrationErrorChannelResolverImpl(BinderAwareChannelResolver resolver,
-                                               IntegrationErrorDestinationBuilder integrationErrorDestinationBuilder) {
-        this.resolver = resolver;
+    public IntegrationErrorChannelResolverImpl(IntegrationErrorDestinationBuilder integrationErrorDestinationBuilder) {
         this.integrationErrorDestinationBuilder = integrationErrorDestinationBuilder;
     }
 
     @Override
-    public MessageChannel resolveDestination(IntegrationRequest event) {
+    public String resolveDestination(IntegrationRequest event) {
         String destination = integrationErrorDestinationBuilder.buildDestination(event);
 
-        return resolver.resolveDestination(destination);
+        return destination;
     }
 
 

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/channels/IntegrationErrorSenderImpl.java

@@ -17,23 +17,26 @@
 
 import org.activiti.cloud.api.process.model.IntegrationError;
 import org.activiti.cloud.api.process.model.IntegrationRequest;
+import org.springframework.cloud.stream.function.StreamBridge;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 
 public class IntegrationErrorSenderImpl implements IntegrationErrorSender {
 
+    private final StreamBridge streamBridge;
     private final IntegrationErrorChannelResolver resolver;
 
-    public IntegrationErrorSenderImpl(IntegrationErrorChannelResolver resolver) {
+    public IntegrationErrorSenderImpl(StreamBridge streamBridge, IntegrationErrorChannelResolver resolver) {
+        this.streamBridge = streamBridge;
         this.resolver = resolver;
     }
 
     @Override
     public void send(Message<IntegrationError> message) {
         IntegrationRequest request = message.getPayload().getIntegrationRequest();
 
-        MessageChannel destination = resolver.resolveDestination(request);
+        String destination = resolver.resolveDestination(request);
 
-        destination.send(message);
+        streamBridge.send(destination, message);
     }
 }

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/channels/IntegrationRequestErrorChannelListener.java

@@ -15,21 +15,21 @@
  */
 package org.activiti.cloud.connectors.starter.channels;
 
-import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.messaging.support.ErrorMessage;
 
 public class IntegrationRequestErrorChannelListener {
 
-    private static final String ERROR_CHANNEL = "errorChannel";
+    public static final String ERROR_CHANNEL = "errorChannel";
 
     private final IntegrationErrorHandler integrationErrorHandler;
 
     public IntegrationRequestErrorChannelListener(IntegrationErrorHandler integrationErrorSender) {
         this.integrationErrorHandler = integrationErrorSender;
     }
 
-    @StreamListener(ERROR_CHANNEL)
-    public void handleError(ErrorMessage errorMessage) {
+    @ServiceActivator(inputChannel = ERROR_CHANNEL)
+    public void accept(ErrorMessage errorMessage) {
         integrationErrorHandler.handleErrorMessage(errorMessage);
     }
 }

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/channels/IntegrationResultChannelResolver.java

@@ -20,6 +20,6 @@
 
 public interface IntegrationResultChannelResolver {
 
-    MessageChannel resolveDestination(IntegrationRequest event);
+    String resolveDestination(IntegrationRequest event);
 
 }

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/channels/IntegrationResultChannelResolverImpl.java

@@ -16,26 +16,20 @@
 package org.activiti.cloud.connectors.starter.channels;
 
 import org.activiti.cloud.api.process.model.IntegrationRequest;
-import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
-import org.springframework.messaging.MessageChannel;
 
 public class IntegrationResultChannelResolverImpl implements IntegrationResultChannelResolver {
 
-    private final BinderAwareChannelResolver resolver;
-
     private final IntegrationResultDestinationBuilder integrationResultDestinationBuilder;
 
-    public IntegrationResultChannelResolverImpl(BinderAwareChannelResolver resolver,
-                                                IntegrationResultDestinationBuilder integrationResultDestinationBuilder) {
-        this.resolver = resolver;
+    public IntegrationResultChannelResolverImpl(IntegrationResultDestinationBuilder integrationResultDestinationBuilder) {
         this.integrationResultDestinationBuilder = integrationResultDestinationBuilder;
     }
 
     @Override
-    public MessageChannel resolveDestination(IntegrationRequest event) {
+    public String resolveDestination(IntegrationRequest event) {
         String destination = integrationResultDestinationBuilder.buildDestination(event);
 
-        return resolver.resolveDestination(destination);
+        return destination;
     }
 
 

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/channels/IntegrationResultSenderImpl.java

@@ -17,23 +17,27 @@
 
 import org.activiti.cloud.api.process.model.IntegrationRequest;
 import org.activiti.cloud.api.process.model.IntegrationResult;
+import org.springframework.cloud.stream.function.StreamBridge;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 
 public class IntegrationResultSenderImpl implements IntegrationResultSender {
 
+    private final StreamBridge streamBridge;
     private final IntegrationResultChannelResolver resolver;
 
-    public IntegrationResultSenderImpl(IntegrationResultChannelResolver resolver) {
+    public IntegrationResultSenderImpl(StreamBridge streamBridge, IntegrationResultChannelResolver resolver) {
+
+        this.streamBridge = streamBridge;
         this.resolver = resolver;
     }
 
     @Override
     public void send(Message<IntegrationResult> message) {
         IntegrationRequest request = message.getPayload().getIntegrationRequest();
 
-        MessageChannel destination = resolver.resolveDestination(request);
+        String destination = resolver.resolveDestination(request);
 
-        destination.send(message);
+        streamBridge.send(destination, message);
     }
 }

activiti-cloud-connectors/activiti-cloud-starter-connector/src/main/java/org/activiti/cloud/connectors/starter/configuration/ActivitiCloudConnectorAutoConfiguration.java

@@ -16,10 +16,24 @@
 package org.activiti.cloud.connectors.starter.configuration;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.activiti.cloud.connectors.starter.channels.*;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorChannelResolver;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorChannelResolverImpl;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorDestinationBuilder;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorDestinationBuilderImpl;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorHandler;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorHandlerImpl;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorSender;
+import org.activiti.cloud.connectors.starter.channels.IntegrationErrorSenderImpl;
+import org.activiti.cloud.connectors.starter.channels.IntegrationRequestErrorChannelListener;
+import org.activiti.cloud.connectors.starter.channels.IntegrationResultChannelResolver;
+import org.activiti.cloud.connectors.starter.channels.IntegrationResultChannelResolverImpl;
+import org.activiti.cloud.connectors.starter.channels.IntegrationResultDestinationBuilder;
+import org.activiti.cloud.connectors.starter.channels.IntegrationResultDestinationBuilderImpl;
+import org.activiti.cloud.connectors.starter.channels.IntegrationResultSender;
+import org.activiti.cloud.connectors.starter.channels.IntegrationResultSenderImpl;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
+import org.springframework.cloud.stream.function.StreamBridge;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.PropertySource;
@@ -59,15 +73,14 @@ public IntegrationResultDestinationBuilder integrationResultDestinationBuilder(C
 
     @Bean
     @ConditionalOnMissingBean
-    public IntegrationResultChannelResolver integrationResultChannelResolver(BinderAwareChannelResolver resolver,
-                                                                             IntegrationResultDestinationBuilder integrationResultDestinationBuilder) {
-        return new IntegrationResultChannelResolverImpl(resolver, integrationResultDestinationBuilder);
+    public IntegrationResultChannelResolver integrationResultChannelResolver(IntegrationResultDestinationBuilder integrationResultDestinationBuilder) {
+        return new IntegrationResultChannelResolverImpl(integrationResultDestinationBuilder);
     }
 
     @Bean
     @ConditionalOnMissingBean
-    public IntegrationResultSender integrationResultSender(IntegrationResultChannelResolver integrationChannelResolver) {
-        return new IntegrationResultSenderImpl(integrationChannelResolver);
+    public IntegrationResultSender integrationResultSender(StreamBridge streamBridge, IntegrationResultChannelResolver integrationChannelResolver) {
+        return new IntegrationResultSenderImpl(streamBridge, integrationChannelResolver);
     }
 
     @Bean
@@ -78,16 +91,15 @@ public IntegrationErrorDestinationBuilder integrationErrorDestinationBuilder(Con
 
     @Bean
     @ConditionalOnMissingBean
-    public IntegrationErrorChannelResolver integrationErrorChannelResolver(BinderAwareChannelResolver resolver,
-                                                                           IntegrationErrorDestinationBuilder integrationErrorDestinationBuilder) {
-        return new IntegrationErrorChannelResolverImpl(resolver, integrationErrorDestinationBuilder);
+    public IntegrationErrorChannelResolver integrationErrorChannelResolver(IntegrationErrorDestinationBuilder integrationErrorDestinationBuilder) {
+        return new IntegrationErrorChannelResolverImpl(integrationErrorDestinationBuilder);
     }
 
 
     @Bean
     @ConditionalOnMissingBean
-    public IntegrationErrorSender integrationErrorSender(IntegrationErrorChannelResolver integrationChannelResolver) {
-        return new IntegrationErrorSenderImpl(integrationChannelResolver);
+    public IntegrationErrorSender integrationErrorSender(StreamBridge streamBridge, IntegrationErrorChannelResolver integrationChannelResolver) {
+        return new IntegrationErrorSenderImpl(streamBridge, integrationChannelResolver);
     }
 
 }