title: Spring Integration - no response on reply channel
body: <p>This is a follow up question to <a href="https://stackoverflow.com/questions/50535484/spring-integration-executor-channel-using-annotations-code-sample/50537931#50537931">Spring Integration Executor Channel using annotations code sample</a>.</p>
  
  <p>System diagram is attached <a href="https://i.stack.imgur.com/q4lUA.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/q4lUA.png" alt="enter image description here"></a>.</p>
  
  <p>I am trying to test the box highlighted in red by posting a message into 'Common channel' and reading from REPLY_CHANNEL set in the msg.</p>
  
  <p>'Common channel' is a publish subscribe channel.
  REPLY_CHANNEL is a QueueChannel.</p>
  
  <p>Since this is a JUnit test, I have mocked jdbcTemplate, datasource and the Impl to ignore any DB calls.</p>
  
  <p>My issue is:
  <strong><em>When I post a message onto 'Common Channel', I do not receive any message on the REPLY_CHANNEL. The junit keeps waiting for a response.</em></strong></p>
  
  <p>What should I change to get a response on the REPLY_CHANNEL?</p>
  
  <pre><code>@RunWith(SpringRunner.class)
  @SpringBootTest
  @ContextConfiguration(loader = AnnotationConfigContextLoader.class) --------- 1 
  @ActiveProfiles("test")
  public class QueuetoQueueTest {
      @Configuration
      static class ContextConfiguration { ------------------------------------- 2
          @Bean(name = "jdbcTemplate")
          public JdbcTemplate jdbcTemplate() {
              JdbcTemplate jdbcTemplateMock = Mockito.mock(JdbcTemplate.class);
              return jdbcTemplateMock;
          }
  
          @Bean(name = "dataSource")
          public DataSource dataSource() {
              DataSource dataSourceMock = Mockito.mock(DataSource.class);
              return dataSourceMock;
          }
  
          @Bean(name = "entityManager")
          public EntityManager entityManager() {
              EntityManager entityManagerMock = Mockito.mock(EntityManager.class);
              return entityManagerMock;
          }
  
          @Bean(name = "ResponseChannel")
          public QueueChannel getReplyQueueChannel() {
              return new QueueChannel();
          }
  
  //This channel serves as the 'common channel' in the diagram
          @Bean(name = "processRequestSubscribableChannel")
          public MessageChannel getPublishSubscribeChannel() {
              return new PublishSubscribeChannel();
          }
      }
  
      @Mock
      DBStoreDaoImpl dbStoreDaoImpl;
  
      @Test
      public void testDBConnectivity() {
          Assert.assertTrue(true);
      }
  
      @InjectMocks -------------------------------------------------------------- 3
      StoretoDBConfig storetoDBConfig = new StoretoDBConfig();
  
      @Autowired
      @Qualifier("ResponseChannel")
      QueueChannel ResponseChannel;
  
      @Autowired
      @Qualifier("processRequestSubscribableChannel")
      MessageChannel processRequestSubscribableChannel;
  
      @Before
      public void setUp() throws Exception {
          MockitoAnnotations.initMocks(this);
      }
  
      @Test
      public void outboundtoQueueTest() {
          try {
              when(dbStoreDaoImpl.storeToDB(any()))
                      .thenReturn(1); ----------------------------------------------- 4
  
              //create message
              Message message = (Message&lt;String&gt;) MessageBuilder
                      .withPayload("Hello")
                      .setHeader(MessageHeaders.REPLY_CHANNEL, ResponseChannel)
                      .build();
  
              //send message
              processRequestSubscribableChannel.send(message);
              System.out
                      .println("Listening on InstructionResponseHandlertoEventProcessorQueue");
  
              //wait for response on reply channel
              Message&lt;?&gt; response = ResponseChannel.receive(); ----------------------- 5
              System.out.println("***************RECEIVED: "
                      + response.toString());
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  }
  </code></pre>
  
  <ol>
  <li><p>Load 'ContextConfiguration' for JUnit so that DB is not accessed.</p></li>
  <li><p>This is how you load custom configuration in JUnit as per <a href="https://spring.io/blog/2011/06/21/spring-3-1-m2-testing-with-configuration-classes-and-profiles" rel="nofollow noreferrer">https://spring.io/blog/2011/06/21/spring-3-1-m2-testing-with-configuration-classes-and-profiles</a></p></li>
  </ol>
  
  <p>Inside the config class, we mock jdbcTemplate, dataSource, entityManager and define the 'common channel' on which the request is posted and ResponseChannel.</p>
  
  <ol start="3">
  <li><p>Inject jdbcTemplate, dataSource mock into StoretoDBConfig so that the DB is not hit</p></li>
  <li><p>Mock DaoImpl class so that DB calls are ignored</p></li>
  <li><p><em>The test blocks here because there is no response on the REPLY_CHANNEL</em></p></li>
  </ol>
  
  <p>UPDATED CODE:</p>
  
  <pre><code>Code inside 5 (the class that reads from common channel):
  
      @Configuration
      class HandleRequestConfig {
  
          //Common channel - refer diagram
          @Autowired
          PublishSubscribeChannel processRequestSubscribableChannel;
  
          //Step 9 - This channel is used to send queue to the downstream system
          @Autowired
          PublishSubscribeChannel forwardToExternalSystemQueue;
  
          public void handle() {
              IntegrationFlows.from("processRequestSubscribableChannel")          // Read from 'Common channel'
              .wireTap(flow-&gt;flow.handle(msg -&gt; System.out.println("Msg received on processRequestSubscribableChannel"+ msg.getPayload())))
              .handle(RequestProcessor,"validateMessage")                         // Perform custom business logic - no logic for now, return the msg as is   
              .wireTap(flow-&gt;flow.handle(msg -&gt; System.out.println("Msg received on RequestProcessor"+ msg.getPayload())))
              .channel("forwardToExternalSystemQueue");                           // Post to 'Channel to another system' 
              }
  
      }
  
      //Code inside step 8 - 'Custom Business Logic' 
      @Configuration
      class RequestProcessor {
          public Message&lt;?&gt; validateMessage(Message&lt;?&gt; msg) {
              return msg;
          }
      }
  </code></pre>
  
  <p><strong>WHAT I AM TRYING TO ACHIEVE:</strong>
  I have individual junit test cases for the business logic. I am trying to test that when the request is posted into the 'common channel', the response is received on 'channel to another system'.</p>
  
  <p><strong>Why I cannot use the original ApplicationContext:</strong> Because it connects to the DB, and I do not want my JUnit to connect to the DB or use an embedded database. I want any calls to the DB to be ignored.</p>
  
  <p>I have set the reply channel to 'ResponseChannel', shouldn't the 'Custom Business Logic' send its response to 'ResponseChannel'?</p>
  
  <p>If I have to listen on a different channel for the response, I am willing to do so. All I want to test is whether the message I am sending on 'common channel' is received on 'channel to other system'. </p>
  
  <p>UPDATE 2:
  Addressing Artem's questions.
  Thankyou Artem for your suggestions.</p>
  
  <p><strong><em>Is 'HandlerRequestConfig' included in the test configuration?</em></strong> - We cannot directly call the handle() method. Instead I thought if I post on 'processRequestSubscribableChannel', the handle() method inside HandleRequestConfig will be invoked since it listens on the same channel. Is this wrong? How do I test HandleRequestConfig.handle() method then?</p>
  
  <p>I added wiretap to the end of each step in HandleRequestConfig (code updated). I find that none of the wiretap message is printed. This means that the msg I am posting is not even reaching the input channel 'processRequestSubscribableChannel'. What am I doing wrong?</p>
  
  <p>NOTE: I tried removing the 'processRequestSubscribableChannel' bean inside Configuration (so that the actual 'processRequestSubscribableChannel' in the applicationContext is used). I am getting an unsatisfied dependency error - Expected atleast 1 bean with configuration PublishSubscribeChannel.</p>
  
  <p>Update 3: Posted details Artem requested.</p>
  
  <pre><code>@RunWith(SpringRunner.class)
  @SpringBootTest
  public class QueuetoQueueTest  {
  
  //  Step 1 - Mocking jdbcTemplate, dataSource, entityManager so that it doesn't connect to the DB
          @MockBean
          @Qualifier("jdbcTemplate")
          JdbcTemplate jdbcTemplate;
  
          @MockBean
          @Qualifier("dataSource")
          public DataSource dataSource;
  
          @MockBean
          @Qualifier("entityManager")
          public EntityManager entityManager;
  
          @Bean(name = "ResponseChannel")
          public PublishSubscribeChannel getReplyQueueChannel() {
              return new PublishSubscribeChannel();
          }
  
          //Mocking the DB class
          @MockBean
          @Qualifier("dbStoreDaoImpl")
          DBStoreDaoImpl  dbStoreDaoImpl ;
  
  
          //Inject the mock objects created above into the flow that stores data into the DB.
          @InjectMocks
          StoretoDBConfig storetoDBConfig = new StoretoDBConfig();
  
  //Step 2 - Injecting MessageChannel used in the actual ApplicationContext
          @Autowired
          @Qualifier("processRequestSubscribableChannel")
          MessageChannel processRequestSubscribableChannel;
  
          @Before
          public void setUp() throws Exception {
              MockitoAnnotations.initMocks(this);
          }
  
          @Test
          public void outboundtoQueueTest() {
          try {
              when(dbStoreDaoImpl.storeToDB(any()))
                      .thenReturn(1);
  
              //create message
              Message message = (Message&lt;?&gt;) MessageBuilder
                      .withPayload("Hello")
                      .build();
              //send message - this channel is the actual channel used in ApplicationContext
              processRequestSubscribableChannel.send(message);
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  }
  </code></pre>
  
  <h2>ERROR I AM GETTING: The code tries to connect to the DB and throws an error.</h2>
  
  <p><strong>UPDATE 1: Code inside StoretoDBConfig</strong> </p>
  
  <pre><code>@Configuration
  @EnableIntegration
  public class StoretoDBConfig {
      @Autowired
      DataSource dataSource;
  
  /*
   * Below code is irrelevant to our current problem - Including for reference.
   * 
   * storing into DB is delegated to a separate thread.
   *
   *  @Bean
   *  public TaskExecutor taskExecutor() {
   *      return new SimpleAsyncTaskExecutor();
   *  }
   *  
   *  @Bean(name="executorChannelToDB")
   *  public ExecutorChannel outboundRequests() {
   *      return new ExecutorChannel(taskExecutor());
   *  }
   *  @Bean(name = "DBFailureChannel")
   *  public static MessageChannel getFailureChannel() {
   *      return new DirectChannel();
   *  }
   *  private static final Logger logger = Logger
   *              .getLogger(InstructionResponseHandlerOutboundtoDBConfig.class);
  */
      @Bean
      public IntegrationFlow handle() {
      /*
       * Read from 'common channel' - processRequestSubscribableChannel and send to separate thread that stores into DB.
       *
       /
          return IntegrationFlows
                  .from("processRequestSubscribableChannel")
                  .channel("executorChannelToDB").get();
      }
  }
  </code></pre>
  
  <p>CODE THAT STORES INTO DB ON THE SEPARATE THREAD:</p>
  
  <pre><code>@Repository
  public class DBStoreDaoImpl implements DBStoreDao {
      private JdbcTemplate jdbcTemplate;
  
      @Autowired
      public void setJdbcTemplate(DataSource dataSource) {
          this.jdbcTemplate = new JdbcTemplate(dataSource);
      }
  
      @Override
      @Transactional(rollbackFor = Exception.class)
      @ServiceActivator(inputChannel = "executorChannelToDB")
      public void storetoDB(Message&lt;?&gt; msg) throws Exception {
              String insertQuery ="Insert into DBTable(MESSAGE) VALUES(?)";
              jdbcTemplate.update(insertQuery, msg.toString());
      }
  }
  </code></pre>