Spring integration recipient-list-router issue

See also questions close to this topic

• Spring Data Mongodb: json array string to BasicDBObject

  I've created this custom converter:

  @Component
  @WritingConverter
  public class MetadataWriterConverter implements Converter<Metadata, DBObject> {
      @Override
      public DBObject convert(Metadata metadata) {
          DBObject dbObject = new BasicDBObject();
          dbObject.put("name", metadata.getName());
          dbObject.put("metadata", (BasicDBObject) BasicDBObject.parse(reference.getMetadata())); 
          dbObject.removeField("_class");
          return dbObject;
      }
  }
  

  I'm getting this exception:

  Caused by: org.bson.BsonInvalidOperationException: readStartDocument can only be called when CurrentBSONType is DOCUMENT, not when CurrentBSONType is ARRAY.

  The problem is on:

  (BasicDBObject) BasicDBObject.parse(metadata.getMetadata())
  

  the content of metadata.getMetadata is: "[{'departament': 'JUST'}]".

  Metadata class is:

  public class Metadata {
      private String id;
      private String user;
      private String metadata;
  }
  

  The content of metadata field is a json string, I'm trying to convert to BasicDbObject, but the problem appears when this string is an json array: [{},{}].

  Guess:

  Metadata met = new Metadata();
  met.setId("Mdt1");
  met.setUser("user");
  met.setMetadata("[{'departament': 'JUST'}]");
  

  What I want to get is:

  {
    "id": Mdt1,
    "user": "user",
    "metadata": [{"departament": "JUST"}]
  }
  

  Any ideas about how to refactor my converter?
• Consider defining a bean named 'entityManagerFactory' in your configuration.3

  I'm just following some tutorial how to upload documents via spring. So while working on it i faced above issue. Since there are many questions related to same question i refer them and tried everything but none corrected m problem. In advance i don't need to add @Repository for my Repository interface and @EnableJpaRepositories in my Application class as i have use correct package sequence. For more reference refer https://dzone.com/articles/spring-spring-boot-and-component-scan. Please help me out.

  Application class

  @SpringBootApplication
  @ComponentScan("com.project.abhishek")
  public class DocumnetwebApplication {
  
  public static void main(String[] args) {
      SpringApplication.run(DocumnetwebApplication.class, args);
  }}
  

  Controller

  @Controller
  public class DocumentController {
  
  @Autowired
  private DocumentRepository documentRepo;
  
  @RequestMapping("/displayUpload")
  public String displayUpload() {
      return "documentUpload";
  }
  
  @RequestMapping(value="/upload", method=RequestMethod.POST )
  public String upload(@RequestParam("document")MultipartFile multiPartFile, @RequestParam("id")long id) {
  
      Document document = new Document(); 
      document.setId(id);
      document.setName(multiPartFile.getName());
      try {
          document.setData(multiPartFile.getBytes());
      } catch (IOException e) {
          e.printStackTrace();
      }
      documentRepo.save(document);
  
      return "documentUpload";
  }}
  

  Bean

  @Entity
  public class Document {
  
  @Id
  private long id;
  private String name;
  @Lob
  private byte[] data;
  //getters and setters}}
  

  Repository

  public interface DocumentRepository extends JpaRepository<Document, Long> {}
  

  Application properties file

  spring.datasource.url=jdbc:mysql://localhost:3306/mydb
  spring.datasource.username=root
  spring.datasource.password=test
  spring.jpa.properties.hibernate.id.new_generator_mappings=false
  spring.jpa.show-sql=true
  server.port=8083
  spring.mvc.view.prefix=/WEB-INF/jsps/
  spring.mvc.view.suffix=.jsp
  server.servlet.context-path= /documentweb
  

  pom file

  <dependencies>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-data-jpa</artifactId>
      </dependency>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-web</artifactId>
      </dependency>
  
      <dependency>
          <groupId>mysql</groupId>
          <artifactId>mysql-connector-java</artifactId>
          <scope>runtime</scope>
      </dependency>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-test</artifactId>
          <scope>test</scope>
      </dependency>
      <dependency>
          <groupId>org.apache.tomcat.embed</groupId>
          <artifactId>tomcat-embed-jasper</artifactId>
          <version>9.0.12</version>
          <scope>provided</scope>
  
      </dependency>
      <dependency>
          <groupId>javax.servlet</groupId>
          <artifactId>jstl</artifactId>
      </dependency>
  
  </dependencies>
  

  Package Structure

  com.project.abhishek---application class  
  com.project.abhishek.controller  
  com.project.abhishek.entity  
  com.project.abhishek.repository
  
• Spring Framework fails on paths with Windows Drive Letters

  I am using Spring Framework version 4.3.18.RELEASE under Intellij on Windows 7. I am looking at the decompiled code of, for example, the class

  org.springframework.core.io.DefaultResourceLoader
  

  and the method

  public Resource getResource(String location).
  

  In that code it has a test for

  location.startsWith("/")
  

  This fails when the location is an absolute Windows path with a leading drive letter. E.g.

  d:/git/thredds/dap4/d4tests/src/test/resources/
  

  Is there a known alternate specification of such a path that will work?
• Parse JSON using Spring SPEL

  Can someone tell me why this does not work:

  @Test
  public void should_parse_json() {
      Expression expression = new SpelExpressionParser().parseExpression("#jsonPath(get('JsonData'), '$.someData')");
  
      Map<String, Object> data = new HashMap<>();
      data.put("JsonData", "{\"someData\": 100}");
  
      StandardEvaluationContext context = new StandardEvaluationContext(data);
      context.addPropertyAccessor(new JsonPropertyAccessor());
  
      assertThat(expression.getValue(context, Object.class)).isEqualTo(100);
  }
  

  I get error "org.springframework.expression.spel.SpelEvaluationException: EL1006E: Function 'jsonPath' could not be found"

  And I have following jar in classpath:

      <dependency>
          <groupId>com.jayway.jsonpath</groupId>
          <artifactId>json-path</artifactId>
      </dependency>
  

  The SPEL documentation did not help me.
• Too many active consumers for ActiveMQ queue

  My Spring app consumes ActiveMQ queue. There are two approaches possible. Initial part of ActiveMQ integration is the same for the both approaches:

  @Bean
  public ConnectionFactory connectionFactory() {
      return new ActiveMQConnectionFactory();
  }
  
  @Bean
  public Queue notificationQueue() {
      return resolveAvcQueueByJNDIName("java:comp/env/jms/name.not.important.queue");
  }
  

  Single thread approach:

  @Bean
  public IntegrationFlow orderNotify() {
      return IntegrationFlows.from(Jms.inboundAdapter(connectionFactory()).destination(notificationQueue()),
              c -> c.poller(Pollers.fixedDelay(QUEUE_POLLING_INTERVAL_MS)
                                   .errorHandler(e -> logger.error("Can't handle incoming message", e))))
                             .handle(...).get();
  }
  

  But I want to consume messages using the several worker threads, so I refactored the code from Inbound adapter to Message driven channel adapter:

  @Bean
  public IntegrationFlow orderNotify() {
      return IntegrationFlows.from(Jms.messageDriverChannelAdapter(connectionFactory()).configureListenerContainer(c -> {
                  final DefaultMessageListenerContainer container = c.get();
                  container.setMaxConcurrentConsumers(notifyThreadPoolSize);
              }).destination(notificationQueue()))
                             .handle(...).get();
  }
  

  The problem is that the app doesn't stop ActiveMQ's consumer when it being redeployed into Tomcat or being restarted for the second approach. It creates new consumer during it's startup. But all new messages are being routed to the old "dead" consumer so they're sat in the "Pending messages" section and never being dequeued.

  What can be the problem here?
• Advice needed on unit testing TCP outbound gateway

  I have a TCP outbound gateway with a custom serializer/deserializer.

  The serialization I assume I can test be setting mock implementations on the properties of the gateway.

  Do I need to have an inbound tcp server gateway to simulate a server?

  Is there a way to mock timeouts, failure to connect and response message without the outbound gateway actually sending data over a socket?

  The goal of the unit test is to explore how error and exception handling is performed in the gateway and as a result understand what the contract with the error channel is.

  Thanks, Alwyn
• Spring JMS unit test produces unexpected result

  I'm trying to determine what happens when I'm using Spring JMS and the broker is down. For that I wrote a unit test like the one below.

  Although I expected the test to fail, it actually succeeds. So how can this happen? The BrokerService is stopped, but stil the JmsTemplate receives a message. Is Spring buffering messages internally? Or is the queue created locally and so it all works even with the broker down?

  @RunWith(SpringRunner.class)
  @ContextConfiguration(classes = JmsConfiguration.class)
  @TestPropertySource("classpath:application.properties")
  public class JmsConfigurationTest
  {
      private static BrokerService broker;
  
      @Autowired
      private Environment env;
  
      @Before
      public void before() throws Exception
      {
          if (broker == null)
          {
              broker = new BrokerService();
              broker.setBrokerName(env.getProperty("activemq.connection.url"));
              broker.setPersistent(false);
              broker.setUseShutdownHook(false);
              broker.start();
          }
      }
  
      @AfterClass
      public static void afterClass() throws Exception
      {
          broker.stop();
      }
  
      @Autowired
      private MessageSender messageSender;
  
      @Autowired
      private JmsTemplate jmsTemplate;
  
      @Test
      public void testSendTestMessageBrokerNotRunning() throws Exception
      {
          broker.stop();
          broker.waitUntilStopped();
  
          messageSender.sendQueue("testrecv", "test1", String.class, "testsender");
  
  
          final String bla = this.jmsTemplate.receiveAndConvert("testrecv").toString();
          assertEquals(bla, "test1");
          broker.start();
      }
  }
  

  The JmsConfiguration class creates the connectionfactory and the jmstemplate by the way.
• How to send messages back from ServiceActivator to Queue in Spring integration

  Hi i have a requirement to launch the spring batch job when there is a message in to the queue and once processing done i need to return some message back to Queue . I am using Spring integration and in Service activator i am able to receive the message from input channel and i am using JmsTemplate to send back response to queue once processing is done. Is this the right way to do it in Spring integration

  @Autowired
  private JmsOperations jmsOperations;
  @ServiceActivator(inputChannel = "inputChannel")
  public void receive(Message<?> message) {
      String content= (String)message.getPayload();
      JobParameters params = new JobParametersBuilder().addString("fileId", content).addString("additionalData", content).toJobParameters();
      try {
          JobExecution jobExecution = jobLauncher.run(job, params);
          if (BatchStatus.COMPLETED.equals(jobExecution.getStatus())) {
          jmsOperations.convertAndSend(jmsSettings.getMatchCompQ(), params.getString("fileId"));                
      }
  

  Thanks in Advance .
• Spring integration org.springframework.integration.MessageTimeoutException handling

  I have following spring integration flow:

          @Bean
          public IntegrationFlow innerInFlow(@Value("${jms.concurrency:10}") String concurrency) {
              return IntegrationFlows
                      .from(Jms.messageDrivenChannelAdapter(
                              Jms.container(connectionFactory, innerQueue)
                                      .concurrency(concurrency)
                                      .taskExecutor(taskExecutor()).get())
                              .extractPayload(true))
                      .transform(Transformers.deserializer())
                      .route(eventRouter())
                      .get();
          }
  

  And after routing

          @Bean
          public IntegrationFlow findPersonClienFlow(FindClientHandler findClientHandler) {
              return IntegrationFlows.from(findPersonClienChannel())
                      .transform(findClientHandler, "queryToFindClientRequest")
                      .handle(Jms.outboundGateway(connectionFactory).requestDestination(cifRequestQueue)
                              .replyDestination(cifResponseQueue).get())
                      .get();
          }
  
      }
  

  In the Jms.outboundGateway I have org.springframework.integration.MessageTimeoutException and I cant understand how I can handle this error?

  Thank you.
• How to call operations on JMS temporary queue from JBOSS CLI

  I am trying to call operations such as listDeliveringMessages() on a temporary Queue (say fe517553-6c53-42d6-8aaa-d8ea154fd8b0), created by a jms.Topic in a JBOSS env.

  I need to do that from JBOSS Cli. I got close to do that by running:

  /subsystem=messaging/hornetq-server=default/jms-queue=fe517553-6c53-42d6-8aaa-d8ea154fd8b0
  

  But the JBOSS cli says that the queue is not found. Yet, If i approach this from JVisualVM and check the MBeans, I can run the operations visually (which is not what I am looking for!):

  

  And when selecting that I can have a list of all attributes:

  

  What command should I run to get the same attributes and call operations from JBOSS cli?
• createdurablesubscriber not receiving persistence messages activemq

  My Reciever is not recieving persistance message. If reciever is online and if I am sending message then it is recieving message but if I am not online (not connect to activeMQ) then I send message after sending message I connect to activeMQ. In this condition I am not getting message. Here is my reciever code.

  ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(Constants.USER_NAME,
              Constants.PASSWORD,Constants.BROKER_URL);
      connectionFactory.setTrustAllPackages(true);
  
  
  
          connection = connectionFactory.createConnection();
          connection.setClientID(Utility.getIMEINumber(context));
          connection.start();
          Topic destination = new ActiveMQTopic(Constants.TOPIC_NAME);
  
          Session receiverSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
          receiverSession.createDurableSubscriber(destination,Utility.getIMEINumber(context));
          MessageConsumer receiver = receiverSession.createConsumer(destination);
  

  My sender code is below

  jmsTemplateTopic.convertAndSend(objectMessage);
  

  the jms topic is configured with persistance delivery
• Jmeter JMS Publisher Issue

  I am trying pushing messgae through jmeter JMS publisher in solace queue. But I am getting below error. Error: javax.jms.JMSException: Error creating connection - internal error (403: Client Username Is Shutdown)

  Also, please confirm on "Destination" field. Currently I am using "Topic JNDI" Value in Destination.
• Only one message get dequeued when configuring JmsTemplate with SESSION_TRANSACTED

  I am trying to configure JmsTemplate in transactional manner, so when error occures durring the processing or writing, i want the messages to be visible in the queue again. I have the below configurations which seems to works fine when error occures i.e. if error is thrown durring the writing or processing, the messages are "sent" back to the queue. But the problem is when everything is OK (i.e. no errors are thrown), then only one message is consumed, processed and dequeued. So if i have 10 messages in the queue, only one is dequeued and the job is marked as COMPLETED. If i configure JmsTemplate with

  jmsTemplate.setSessionTransacted(false);
  

  I have all messages consumed, processed and written, but of course I am loosing the transactional manner. The question is how to get all mesasges transactionaly consumed and processed and written?

  I have the following JmsTemplate configuration

  @Bean
  public JmsTemplate jmsTemplate() {
      JmsTemplate jmsTemplate = new JmsTemplate(activeMQConnectionFactory());
      jmsTemplate.setDefaultDestinationName(applicationProperties.getQueue());
      jmsTemplate.setDefaultDestination(queue());
      jmsTemplate.setReceiveTimeout(applicationProperties.getReceiveTimeout());
      jmsTemplate.setSessionTransacted(true);
      jmsTemplate.setSessionAcknowledgeMode(Session.SESSION_TRANSACTED);
      return jmsTemplate;
  }
  

  which is passed to my JmsItemReader

  @Bean
  public ItemReader<MyPojoClass> myJmsItemReader() throws Exception {
  
      JmsItemReader<MyPojoClass> myJmsItemReader= new JmsItemReader<>();
      myJmsItemReader.setItemType(MyPojoClass.class);
      myJmsItemReader.setJmsTemplate(jmsTemplate);
      myJmsItemReader.afterPropertiesSet();
      return myJmsItemReader;
  
  }
  

  and below is my tasklet configuration

  @Bean(name = "receiveProcessAndWrite")
  public TaskletStep receiveProcessAndWrite() throws Exception {
  
      return this.stepBuilderFactory.get("receiveProcessAndWrite")
              .transactionManager(transactionManager)
              .<MyPojoClass, MyPojoClass>chunk(1000)
              .readerIsTransactionalQueue()
              .reader(myJmsItemReader())
              .processor(myprocessor())
              .writer(updateStatusInDBandWrite())
              .listener(new ChunkCountListener())
              .build();
  
  }
  
• JmsTemplate with RECEIVE_TIMEOUT_NO_WAIT doesn't retrieve messages from JMS queue

  Given I have ActiveMQ queue where many messages are already present.

  When I set receive timeout on JmsTemplate to RECEIVE_TIMEOUT_NO_WAIT which is equal to -1:

  jmsTemplate.setReceiveTimeout(JmsTemplate.RECEIVE_TIMEOUT_NO_WAIT); 
  

  and try to receive one of those messages:

  Message msg = jmsTemplate.receive(queueName);
  

  then msg is null, but it should not be according JavaDoc:

  /**
   * Timeout value indicating that a receive operation should
   * check if a message is immediately available without blocking.
   */
  public static final long RECEIVE_TIMEOUT_NO_WAIT = -1;
  

  Why is that?

  When I do:

  jmsTemplate.setReceiveTimeout(1000);
  

  then messages are retrieved.