0 Replies Latest reply on Nov 3, 2008 4:57 AM by Jayasree B

    Urgent -MultiplexingConsumerProcessor Error while handling jms message

    Jayasree B Novice

      Hi,

       

      I am using fuse-esb-3.3.1.3 and fuse-message-broker-5.0.0.17.

      My requirrement is to build a cxf-se SU which has wsdl and implementation class.I have servicemix-jms SU which acts as consumer and picks up the message from queue.

       

      SOAp message is converted to JMS Message and placed it an queue by a client program. This message should  be picked by servicemix-jms and it should send it to cxf-se for further processing.

       

      I have service engine service unit which follows WSDL first approach .

       

      After compilation and deployment of the project , when i try to place in the message in Queue using client program i am gettting following error.

       

      It would be very helpful if I get guidance to solve this.

       

      My Client program to place JMS Message is

       

      package com.test;

       

       

      import javax.xml.bind.JAXBContext;

      import javax.xml.bind.Marshaller;

       

      import javax.xml.soap.MessageFactory;

       

       

      import javax.jms.Connection;

      import javax.jms.DeliveryMode;

      import javax.jms.Destination;

      import javax.jms.Message;

      import javax.jms.MessageProducer;

      import javax.jms.Session;

       

       

      import javax.xml.soap.SOAPBody;

       

      import javax.xml.soap.SOAPEnvelope;

      import javax.xml.soap.SOAPMessage;

      import javax.xml.soap.SOAPPart;

       

       

       

       

      import org.apache.activemq.ActiveMQConnection;

      import org.apache.activemq.ActiveMQConnectionFactory;

      import org.apache.activemq.util.IndentPrinter;

       

       

       

       

      import com.sun.messaging.xml.MessageTransformer;

      import com.cvs.eph.entities._1_0.*;

      import com.cvs.eph.messages._1_0.*;

       

      public class Producer {

            

           private Destination destination;

         // private int messageCount = 10;

          private long sleepTime;

          //private boolean verbose = true;

          private int messageSize = 255;

          private long timeToLive;

          private String user = ActiveMQConnection.DEFAULT_USER;

          private String password = ActiveMQConnection.DEFAULT_PASSWORD;

          private String url = ActiveMQConnection.DEFAULT_BROKER_URL;

          //private String subject = "TOOL1.DEFAULT";

          private String subject = "queues/EPHOngoingMaintenanceInput";

          private boolean topic;

          private boolean transacted;

          private boolean persistent;

          public static void main(String[] args) {

              Producer producerTool = new Producer();

              producerTool.run();

          }

          public void run() {

              Connection connection = null;

              try {

                  System.out.println("Connecting to URL: " + url);

                  System.out.println("Publishing a Message with size " + messageSize + " to " + (topic ? "topic" : "queue") + ": " + subject);

                  System.out.println("Using " + (persistent ? "persistent" : "non-persistent") + " messages");

                  System.out.println("Sleeping between publish " + sleepTime + " ms");

                  if (timeToLive != 0) {

                      System.out.println("Messages time to live " + timeToLive + " ms");

                  }

       

                  // Create the connection.

               ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);

                //  ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://blrkec38454d.ad.infosys.com:61616");

                   

                  connection = connectionFactory.createConnection();

                  connection.start();

       

                  // Create the session

                  Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);

                  if (topic) {

                      destination = session.createTopic(subject);

                  } else {

                      destination = session.createQueue(subject);

                  }

       

                  // Create the producer.

                  MessageProducer producer = session.createProducer(destination);

                  if (persistent) {

                      producer.setDeliveryMode(DeliveryMode.PERSISTENT);

                  } else {

                      producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

                  }

                  if (timeToLive != 0) {

                      producer.setTimeToLive(timeToLive);

                  }

       

                  // Start sending messages

                  sendLoop(session, producer);

       

                  System.out.println("Done.");

       

                  // Use the ActiveMQConnection interface to dump the connection

                  // stats.

                  ActiveMQConnection c = (ActiveMQConnection)connection;

                  c.getConnectionStats().dump(new IndentPrinter());

       

              } catch (Exception e) {

                  System.out.println("Caught: " + e);

                  e.printStackTrace();

              } finally {

                  try {

                      connection.close();

                  } catch (Throwable ignore) {

                  }

              }

          }

          protected void sendLoop(Session session, MessageProducer producer) throws Exception {

       

              //for (int i = 0; i

       

       

      I am getting following error when i place message in queue......

       

      ERROR - MultiplexingConsumerProcessor  - Error while handling jms message

      com.ctc.wstx.exc.WstxIOException: Invalid UTF-8 middle byte 0xed (at char #3, by

      te #-1)

              at com.ctc.wstx.sr.StreamScanner.throwFromIOE(StreamScanner.java:683)

              at com.ctc.wstx.sr.BasicStreamReader.next(BasicStreamReader.java:1086)

              at javax.xml.stream.util.StreamReaderDelegate.next(StreamReaderDelegate.

      java:59)

              at org.apache.servicemix.jbi.jaxp.ExtendedXMLStreamReader.next(ExtendedX

      MLStreamReader.java:61)

              at org.apache.servicemix.jbi.jaxp.ExtendedXMLStreamReader.nextTag(Extend

      edXMLStreamReader.java:44)

              at org.apache.servicemix.soap.marshalers.SoapReader.readSoapUsingStax(So

      apReader.java:164)

              at org.apache.servicemix.soap.marshalers.SoapReader.read(SoapReader.java

      :89)

              at org.apache.servicemix.soap.marshalers.SoapReader.read(SoapReader.java

      :76)

              at org.apache.servicemix.jms.DefaultJmsMarshaler.toSOAP(DefaultJmsMarsha

      ler.java:179)

              at org.apache.servicemix.jms.AbstractJmsProcessor.toNMS(AbstractJmsProce

      ssor.java:168)

              at org.apache.servicemix.jms.multiplexing.MultiplexingConsumerProcessor.

      access$300(MultiplexingConsumerProcessor.java:38)

              at org.apache.servicemix.jms.multiplexing.MultiplexingConsumerProcessor$

      1.run(MultiplexingConsumerProcessor.java:89)

              at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExec

      utor.java:650)

              at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor

      .java:675)

              at java.lang.Thread.run(Thread.java:595)

      Caused by: java.io.CharConversionException: Invalid UTF-8 middle byte 0xed (at c

      har #3, byte #-1)

              at com.ctc.wstx.io.UTF8Reader.reportInvalidOther(UTF8Reader.java:310)

              at com.ctc.wstx.io.UTF8Reader.read(UTF8Reader.java:208)

              at com.ctc.wstx.io.ReaderSource.readInto(ReaderSource.java:84)

              at com.ctc.wstx.io.BranchingReaderSource.readInto(BranchingReaderSource.

      java:57)

              at com.ctc.wstx.sr.StreamScanner.loadMore(StreamScanner.java:967)

              at com.ctc.wstx.sr.StreamScanner.getNext(StreamScanner.java:738)

              at com.ctc.wstx.sr.BasicStreamReader.nextFromProlog(BasicStreamReader.ja

      va:1995)

              at com.ctc.wstx.sr.BasicStreamReader.next(BasicStreamReader.java:1069)

              ... 13 more

       

       

       

      I have uploaded my project also. Pls help me to resolve this.

       

      Jayasree.B