JMS with akka and multi threading

Question

public class QueueListener implements MessageListener {

    public static final ExecutorService executor = Executors.newWorkStealingPool();

    public static boolean isActorinit=false;
    public static ActorSystem system=null;
    private ActorRef myActor=null;
    private String _queueName=null; 

    public QueueListener(String qName){
        this._queueName = qName;
        if(!isActorinit){
            system=ActorSystem.create("Controller");

            try {
            myActor=system.actorOf(Props.create(MessageExecutor.class.getConstructor(String.class).newInstance(_queueName).getClass()),"mysysActor");
            } catch (Exception e) {
                // TODO Auto-generated catch block
            }
            isActorinit=true;
        }
    }

    /* 
     * (non-Javadoc)
     * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
     */
    @Override
    public void onMessage(Message msg) {

//      processRequest(msg);
        executeRequest(msg);
    }

    /** This method will process the message fetch by the listener.
     *   
     * @param msg - javax.jms.Messages parameter get queue message
     */
    private void processRequest(Message msg){

        String requestData=null;
        try {

            if(msg instanceof TextMessage){
                TextMessage textMessage= (TextMessage) msg;
                requestData = textMessage.getText().toString();
            }else if(msg instanceof ObjectMessage){
                ObjectMessage objMsg = (ObjectMessage) msg; 
                requestData = objMsg.getObject().toString();
            }


            MessageProcessor msgProcessor = new MessageProcessor(_queueName, requestData);
            executor.submit(msgProcessor);
        } catch (JMSException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }

    }

    private void executeRequest(Message msg){

        String requestData=null;
        try {

            if(msg instanceof TextMessage){
                TextMessage textMessage= (TextMessage) msg;
                requestData = textMessage.getText().toString();
            }else if(msg instanceof ObjectMessage){
                ObjectMessage objMsg = (ObjectMessage) msg; 
                requestData = objMsg.getObject().toString();
            }
//           MessageExecutor objMessageExecutor=new MessageExecutor(_queueName);
            myActor.tell(requestData, ActorRef.noSender()); 

        } catch (JMSException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }

    }

}

This code is working fine when executing ProcessRequst method with ExecutorService. however, facing below issue with akka actor system implementation.

Exception in thread "Thread-4" java.lang.NullPointerException
    at com.syn.jms.listener.QueueListener.executeRequest(QueueListener.java:102)
    at com.syn.jms.listener.QueueListener.onMessage(QueueListener.java:59)
    at org.apache.qpid.amqp_1_0.jms.impl.SessionImpl$Dispatcher.run(SessionImpl.java:942)
    at java.lang.Thread.run(Thread.java:745)

Please note that I am using Apache qpid APi for AMQP protocol with activeMQ.

I am unable to understand the issue.

Answer 1

I have found the solution it was due to NPE of Actor ref while processing for multiple inputs with unique actorRef for each process queue and it was not initializing object. This solution I have found.

public QueueListener(String actorId,String qName){
        this._queueName = qName;
         if(!isActorinit){
                system=ActorSystem.create(actorId);

                isActorinit=true;
            }

          myActor=system.actorOf( Props.create(MessageExecutor.class, qName),qName);
    }

However, I appreaciate your input to reach me to the solution. Thnaks