Transactional stream binder (#188)

* Stream Binder transactions

Signed-off-by: Anders Swanson <anders.swanson@oracle.com>

Author: anders-swanson

database/spring-cloud-stream-binder-oracle-txeventq/spring-cloud-stream-binder-txeventq-sample/src/test/java/com/oracle/database/spring/cloud/stream/binder/sample/TxEventQSampleAppTest.java

@@ -20,7 +20,7 @@
 @Testcontainers
 public class TxEventQSampleAppTest {
     @Container
-    static OracleContainer oracleContainer = new OracleContainer("gvenzl/oracle-free:23.6-slim-faststart")
+    static OracleContainer oracleContainer = new OracleContainer("gvenzl/oracle-free:23.7-slim-faststart")
             .withStartupTimeout(Duration.ofMinutes(2))
             .withUsername("testuser")
             .withPassword("testpwd");

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/JMSMessageChannelBinder.java

@@ -30,16 +30,13 @@
 import com.oracle.database.spring.cloud.stream.binder.config.JmsProducerProperties;
 import com.oracle.database.spring.cloud.stream.binder.provisioning.JmsConsumerDestination;
 import com.oracle.database.spring.cloud.stream.binder.provisioning.JmsProducerDestination;
-
-
 import com.oracle.database.spring.cloud.stream.binder.utils.DestinationNameResolver;
 import com.oracle.database.spring.cloud.stream.binder.utils.JmsMessageDrivenChannelAdapterFactory;
 import com.oracle.database.spring.cloud.stream.binder.utils.JmsSendingMessageHandlerFactory;
 import jakarta.jms.Connection;
 import jakarta.jms.ConnectionFactory;
 import jakarta.jms.Session;
 import jakarta.jms.Topic;
-
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/TxEventQQueueProvisioner.java

@@ -1,6 +1,6 @@
 /*
  ** TxEventQ Support for Spring Cloud Stream
- ** Copyright (c) 2023, 2024 Oracle and/or its affiliates.
+ ** Copyright (c) 2023, 2025 Oracle and/or its affiliates.
  **
  ** This file has been modified by Oracle Corporation.
  */
@@ -24,6 +24,8 @@
 
 package com.oracle.database.spring.cloud.stream.binder;
 
+import java.sql.SQLException;
+
 import com.oracle.database.spring.cloud.stream.binder.config.JmsConsumerProperties;
 import com.oracle.database.spring.cloud.stream.binder.config.JmsProducerProperties;
 import com.oracle.database.spring.cloud.stream.binder.plsql.OracleDBUtils;
@@ -34,9 +36,6 @@
 import jakarta.jms.JMSException;
 import jakarta.jms.Session;
 import jakarta.jms.Topic;
-
-import java.sql.SQLException;
-
 import oracle.jakarta.jms.AQjmsException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -137,7 +136,7 @@ private Topic provisionProducerTopic(String topicName,
                                          ExtendedProducerProperties<JmsProducerProperties> properties) {
         Connection aQConnection = null;
         Session session = null;
-        Topic topic = null;
+        Topic topic;
         try {
             aQConnection = connectionFactory.createConnection();
             session = aQConnection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
@@ -177,7 +176,7 @@ private Topic provisionConsumerTopic(String topicName,
                                          ExtendedConsumerProperties<JmsConsumerProperties> properties) {
         Connection aQConnection = null;
         Session session = null;
-        Topic topic = null;
+        Topic topic;
         try {
             aQConnection = connectionFactory.createConnection();
             session = aQConnection.createSession(true, Session.CLIENT_ACKNOWLEDGE);

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/config/JmsBinderGlobalConfiguration.java

@@ -46,7 +46,7 @@
 @Configuration
 public class JmsBinderGlobalConfiguration {
 
-    private final ConnectionFactory connectionFactory;
+    private ConnectionFactory connectionFactory;
 
     public JmsBinderGlobalConfiguration(ConnectionFactory connectionFactory) {
         this.connectionFactory = connectionFactory;

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/config/JmsExtendedBindingProperties.java

@@ -27,8 +27,8 @@
 import java.util.Map;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
 import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
 
 @ConfigurationProperties("spring.cloud.stream.txeventq")
 public class JmsExtendedBindingProperties

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/config/TxEventQJmsConfiguration.java

@@ -24,17 +24,15 @@
 
 package com.oracle.database.spring.cloud.stream.binder.config;
 
+import java.sql.SQLException;
+
 import com.oracle.database.spring.cloud.stream.binder.TxEventQQueueProvisioner;
 import com.oracle.database.spring.cloud.stream.binder.plsql.OracleDBUtils;
-
 import jakarta.jms.ConnectionFactory;
 import jakarta.jms.JMSException;
 import oracle.jakarta.jms.AQjmsConnectionFactory;
 import oracle.jakarta.jms.AQjmsFactory;
 import oracle.ucp.jdbc.PoolDataSource;
-
-import java.sql.SQLException;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/plsql/OracleDBUtils.java

@@ -29,15 +29,14 @@
 import java.sql.SQLException;
 import java.sql.Types;
 
+import oracle.ucp.jdbc.PoolDataSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import oracle.ucp.jdbc.PoolDataSource;
-
 public class OracleDBUtils {
 
     private PoolDataSource pds = null;
-    private final int dbversion;
+    private int dbversion;
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     private static final String CREATE_KB2_TEQ =

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/provisioning/JmsProducerDestination.java

@@ -34,7 +34,7 @@ public class JmsProducerDestination implements ProducerDestination {
 
     private final Topic topic;
     private final int partitionCount;
-    private final int dbversion;
+    private int dbversion;
 
     public JmsProducerDestination(Topic topic, int pCount, int dbversion) {
         this.topic = topic;

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/serialize/CustomSerializationMessageConverter.java

@@ -1,6 +1,6 @@
 /*
  ** TxEventQ Support for Spring Cloud Stream
- ** Copyright (c) 2023, 2024 Oracle and/or its affiliates.
+ ** Copyright (c) 2023, 2025 Oracle and/or its affiliates.
  **
  ** This file has been modified by Oracle Corporation.
  */
@@ -24,13 +24,12 @@
 
 package com.oracle.database.spring.cloud.stream.binder.serialize;
 
+import jakarta.jms.JMSException;
+import jakarta.jms.Message;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.jms.support.converter.SimpleMessageConverter;
 
-import jakarta.jms.JMSException;
-import jakarta.jms.Message;
-
 public class CustomSerializationMessageConverter extends SimpleMessageConverter {
     public String deserializer = null;
 
@@ -66,8 +65,8 @@ public Object fromMessage(Message jmsMessage) throws JMSException {
         }
 
         if (!isInstanceOfDeserializer) {
-            logger.debug("The configured deserializer class is not an instance of 'com.oracle.database.spring.cloud.stream.binder.serialize.DeSerializer'");
-            throw new IllegalArgumentException("The configured serializer class is not an instance of 'com.oracle.database.spring.cloud.stream.binder.serialize.DeSerializer'");
+            logger.debug("The configured deserializer class is not an instance of 'com.oracle.cstream.serialize.DeSerializer'");
+            throw new IllegalArgumentException("The configured serializer class is not an instance of 'com.oracle.cstream.serialize.DeSerializer'");
         }
 
         Deserializer<?> s = null;
@@ -79,7 +78,7 @@ public Object fromMessage(Message jmsMessage) throws JMSException {
             throw new IllegalArgumentException("Serializer object could not be initiated.");
         }
 
-        result = s.deserialize((byte[]) result);
+        result = (Object) (s.deserialize((byte[]) result));
 
         return result;
     }

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/serialize/Deserializer.java

@@ -25,5 +25,5 @@
 package com.oracle.database.spring.cloud.stream.binder.serialize;
 
 public interface Deserializer<T> {
-    T deserialize(byte[] bytes);
+    public T deserialize(byte[] bytes);
 }

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/serialize/Serializer.java

@@ -25,5 +25,5 @@
 package com.oracle.database.spring.cloud.stream.binder.serialize;
 
 public interface Serializer {
-    byte[] serialize(Object data);
+    public byte[] serialize(Object data);
 }

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/utils/DestinationNameResolver.java

@@ -27,7 +27,7 @@
 import org.springframework.util.StringUtils;
 
 public class DestinationNameResolver {
-    private final AnonymousNamingStrategy namingStrategy;
+    private AnonymousNamingStrategy namingStrategy;
 
     public DestinationNameResolver(AnonymousNamingStrategy namingStrategy) {
         this.namingStrategy = namingStrategy;

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/utils/JmsMessageDrivenChannelAdapterFactory.java

@@ -26,13 +26,12 @@
 
 import com.oracle.database.spring.cloud.stream.binder.config.JmsConsumerProperties;
 import com.oracle.database.spring.cloud.stream.binder.serialize.CustomSerializationMessageConverter;
-
 import jakarta.jms.BytesMessage;
 import jakarta.jms.Destination;
 import jakarta.jms.JMSException;
 import jakarta.jms.Message;
 import jakarta.jms.Session;
-
+import oracle.jakarta.jms.AQjmsSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
@@ -60,7 +59,7 @@ public class JmsMessageDrivenChannelAdapterFactory
 
     private ApplicationContext applicationContext;
 
-    private final Logger logger = LoggerFactory.getLogger(getClass());
+    private Logger logger = LoggerFactory.getLogger(getClass());
 
     public JmsMessageDrivenChannelAdapterFactory(
             ListenerContainerFactory listenerContainerFactory,
@@ -145,10 +144,12 @@ private static class RetryingChannelPublishingJmsMessageListener
 
         private final RetryTemplate retryTemplate;
 
-        private final RecoveryCallback<Object> recoverer;
+        private RecoveryCallback<Object> recoverer;
 
         private String deSerializerClassName = null;
 
+        SpecCompliantJmsHeaderMapper headerMapper = new SpecCompliantJmsHeaderMapper();
+
         RetryingChannelPublishingJmsMessageListener(
                 ConsumerProperties properties,
                 MessageRecoverer messageRecoverer,
@@ -191,6 +192,9 @@ public Object doWithRetry(RetryContext retryContext)
                                                 RETRY_CONTEXT_MESSAGE_ATTRIBUTE,
                                                 jmsMessage
                                         );
+
+                                        headerMapper.setConnection(((AQjmsSession) session).getDBConnection());
+                                        RetryingChannelPublishingJmsMessageListener.super.setHeaderMapper(headerMapper);
                                         RetryingChannelPublishingJmsMessageListener.super.onMessage(
                                                 jmsMessage,
                                                 session

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/utils/JmsSendingMessageHandlerFactory.java

@@ -24,6 +24,7 @@
 
 package com.oracle.database.spring.cloud.stream.binder.utils;
 
+import jakarta.jms.Destination;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -33,8 +34,6 @@
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.messaging.MessageChannel;
 
-import jakarta.jms.Destination;
-
 public class JmsSendingMessageHandlerFactory implements ApplicationContextAware, BeanFactoryAware {
 
     private final JmsTemplate template;

database/spring-cloud-stream-binder-oracle-txeventq/src/main/java/com/oracle/database/spring/cloud/stream/binder/utils/ListenerContainerFactory.java

@@ -26,14 +26,14 @@
 
 import jakarta.jms.ConnectionFactory;
 import jakarta.jms.Destination;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
 import org.springframework.jms.listener.DefaultMessageListenerContainer;
 
 public class ListenerContainerFactory {
-    private final ConnectionFactory factory;
+
+    private ConnectionFactory factory;
 
     private static final Logger logger = LoggerFactory.getLogger(ListenerContainerFactory.class);