0

Communication of 2 Java SWT Applications via Apache ActiveMQ

Apache ActiveMQ  is the most popular and powerful open source messaging, message oriented broker. It supports many cross language clients. You can use ActiveMQ with any language(C, C++, Java..), any protocols (TCP, UDP, Openwire, Rest…). Also you can manage queue and asynchronous messaging.

ActiveMQ has 3 main mechanism:

Broker: It gets message from producers and send it to the consumers. Broker can be used as embedded or ActiveMQ server can be used as broker.

Producer:It creates message and send it to the broker.

Consumer: It gets message from broker and process it.

1

You can also use topic and queue to send message to receiver from sender.

23

In this post, we will learn how to use ActiveMQ with Java SWT Applications. I downloaded  ActiveMQ and installed it on Linux and Windows OS.  You can reach ActiveMQ Installion from this link. You can run ActiveMQ  in linux using this commands.

cd {ActiveMq Directory}/bin

./activemq start   # to start activemq

netstat -an|grep 61616 # to show the 61616 listening port

./activemq stop   # to stop activemq

Also, you can see configuration and details in browser using http://localhost:8161/admin/. Username: admin; password: admin.

I have two  Java SWT application: Module 1 and Module 2. In scenario, Module 2 sents number to Module 1 using ActiveMq.

4

To use code, you should add external jars: javax.jms-3.1.1.jar, openwire-example-0.1-SNAPSHOT.jar.

GitHUB Links:

https://github.com/omerbsezer/Module1

https://github.com/omerbsezer/Module2

Module 1 Receiving Side has 2 part: MainScreen (for gui) and Thread1 (for connecting activemq and receiving data ) .

MainScreen.java

package MainPackage;
import org.eclipse.swt.widgets.Display;
 import org.eclipse.swt.widgets.Shell;
 import org.eclipse.swt.widgets.Label;
 import org.eclipse.swt.SWT;
 import org.eclipse.swt.widgets.Text;
 import org.eclipse.swt.widgets.Group;
public class MainScreen {
    protected Shell shlModule;
     public static Text ReceivingText;
     public static String string_ref_temp=null;
     private Group grpModuleReceiving;
    /**
      * Launch the application.
      * @param args
      */
     public static void main(String[] args) {
         try {
             MainScreen window = new MainScreen();
             window.open();
         } catch (Exception e) {
             e.printStackTrace();
         }
     }
     public static void doUpdate(final Display display, final Text target,
               final String value) {
          
              Display.getDefault().asyncExec(new Runnable() {
               @Override
               public void run() {
                 if (!target.isDisposed()) {
                   target.setText(value);
                   target.getParent().layout();
                 }
               }
             });
           }
     public static void UpdateInfo(final Display display, final Label target,
               final String value) {
          
              Display.getDefault().asyncExec(new Runnable() {
               @Override
               public void run() {
                 if (!target.isDisposed()) {
                   target.setText(value);
                   target.getParent().layout();
                 }
               }
             });
           }
      
     /**
      * Open the window.
      */
     public void open() {
         Display display = Display.getDefault();
         createContents();
         
         Thread1 MThread = new Thread1();
         Thread t = new Thread(MThread);
         t.start();
         
         shlModule.open();
         shlModule.layout();
         while (!shlModule.isDisposed()) {
             if (!display.readAndDispatch()) {
                 display.sleep();
             }
         }
     }
    /**
      * Create contents of the window.
      */
     protected void createContents() {
         shlModule = new Shell();
         shlModule.setSize(450, 198);
         shlModule.setText("Module 1");
         
         grpModuleReceiving = new Group(shlModule, SWT.NONE);
         grpModuleReceiving.setText("Module 1 Receiving Part");
         grpModuleReceiving.setBounds(29, 10, 389, 117);
         
         ReceivingText = new Text(grpModuleReceiving, SWT.BORDER);
         ReceivingText.setBounds(168, 49, 104, 27);
         
         Label lblReceiving = new Label(grpModuleReceiving, SWT.NONE);
         lblReceiving.setBounds(81, 59, 68, 17);
         lblReceiving.setText("Receiving:");
    }
 }

Thread1.java

package MainPackage;
import javax.jms.Connection;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
 import org.eclipse.swt.widgets.Display;
public class Thread1 extends MainScreen  implements Runnable {
     
     public  Destination dest_ref_temp;
     
     
     public  MessageConsumer consumer_ref_temp;
     
     
     
     public  String user=null,password=null,host=null;
     public  String text_ref_temp = null;
     public int port;
     public Session session_ref_temp;
     
     private Display display;
    
     public Display getDisplay(){
         return display;
     }
    @Override
     public void run() {
         // TODO Auto-generated method stub
         user = env("ACTIVEMQ_USER", "admin");
         password = env("ACTIVEMQ_PASSWORD", "password");
         host = env("ACTIVEMQ_HOST", "localhost");
         int port = Integer.parseInt(env("ACTIVEMQ_PORT", "61616"));
         //String destination = arg(args, 0, "event");
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://" + host + ":" + port);
        Connection connection = null;
         try {
             connection = factory.createConnection(user, password);
             //connection = factory.createConnection();
             connection.start();
            
             session_ref_temp = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             dest_ref_temp = session_ref_temp.createQueue("REF_TEMP.FOO");
             consumer_ref_temp = session_ref_temp.createConsumer(dest_ref_temp);
         
         } catch (JMSException e1) {
             // TODO Auto-generated catch block
             e1.printStackTrace();
         }
         
        System.out.println("Waiting for messages...");
         
         while (true) {
             // Wait for a message
             
             Message message_ref_temp = null;
             
             try {
                 message_ref_temp  = consumer_ref_temp.receive(1000);
             } catch (JMSException e1) {
                 // TODO Auto-generated catch block
                 e1.printStackTrace();
             }
              
             if (message_ref_temp == null){
                 continue;
             }
             
              
             if (message_ref_temp instanceof TextMessage ) {
                     TextMessage textMessage_ref_temp = (TextMessage) message_ref_temp;
                   
                     try {
                         text_ref_temp = textMessage_ref_temp.getText();        
                     } catch (JMSException e) {
                         // TODO Auto-generated catch block
                         e.printStackTrace();
                     }
                
                    string_ref_temp=String.valueOf(text_ref_temp);
                    doUpdate(display,ReceivingText,text_ref_temp);
        
                 } else {
              
                 }
         } 
         /*try {
             consumer.close();
             session.close(); 
             connection.close();
         } catch (JMSException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }*/
         
     }
    private static String env(String key, String defaultValue) {
         String rc = System.getenv(key);
         if( rc== null )
             return defaultValue;
         return rc;
     }
 }

Module 2 Sending Side has 2 part: M2MainScreen (for gui) and SendingThread (for connecting activemq and sending data ) .

M2MainScreen.java

package MainPackage;
import org.eclipse.swt.widgets.Display;
 import org.eclipse.swt.widgets.Event;
 import org.eclipse.swt.widgets.Listener;
 import org.eclipse.swt.widgets.Shell;
 import org.eclipse.swt.widgets.Text;
 import org.eclipse.swt.SWT;
 import org.eclipse.swt.widgets.Label;
 import org.eclipse.swt.widgets.Button;
 import org.eclipse.swt.widgets.Group;
 import org.eclipse.swt.widgets.ProgressBar;
public class M2MainScreen {
    protected Shell shlModule;
    
     public static Display display;
     public static Text ref_temp;
     public static Button btnStart;
     public static Button btnStop;
     public static Label lblInfo;
     
     public static String string_ref_temp;
     public static String isRunning;
     public static int i;
     public static int stat=1;
     public static int ref_temp_value=0;
    /**
      * Launch the application.
      * @param args
      */
     public static void main(String[] args) {
         try {
             M2MainScreen window = new M2MainScreen();
             window.open();
         } catch (Exception e) {
             e.printStackTrace();
         }
     }
     public static void doUpdate(final Display display, final Text target,
               final String value) {
          
              Display.getDefault().asyncExec(new Runnable() {
               @Override
               public void run() {
                 if (!target.isDisposed()) {
                   target.setText(value);
                   target.getParent().layout();
                 }
               }
             });
           }
     /**
      * Open the window.
      */
     public void open() {
         Display display = Display.getDefault();
         createContents();
         
         Listener listener = new Listener() {
             public void handleEvent(Event event) {
               if (event.widget == btnStart) {
                 //response[0] = true;
                   
                   lblInfo.setText("Module2 Sending Action Started!");
                   isRunning="true";
                   SendingThread sendingThread = new SendingThread();
                   Thread t = new Thread(sendingThread);
                     t.start();
               } else if (event.widget == btnStop) {
                   lblInfo.setText("Module2 Sending Action Stopped!");
                   isRunning="false";
                   stat=1;
               }
               else{
                   
               }
               //dialog.close();
             }
           };
           
         btnStart.addListener(SWT.Selection, listener);
         btnStop.addListener(SWT.Selection, listener);
         
         shlModule.open();
         shlModule.layout();
         while (!shlModule.isDisposed()) {
             if (!display.readAndDispatch()) {
                 display.sleep();
             }
         }
     }
    /**
      * Create contents of the window.
      */
     protected void createContents() {
         shlModule = new Shell();
         shlModule.setSize(450, 268);
         shlModule.setText("Module 2");
         
         lblInfo = new Label(shlModule, SWT.NONE);
         lblInfo.setBounds(79, 216, 290, 17);
         
         Group grpModuleSending = new Group(shlModule, SWT.NONE);
         grpModuleSending.setText("Module 2 Sending Part");
         grpModuleSending.setBounds(33, 10, 375, 200);
         
         Label lblNewLabel = new Label(grpModuleSending, SWT.NONE);
         lblNewLabel.setBounds(59, 60, 68, 17);
         lblNewLabel.setText("Sending: ");
         
         ref_temp = new Text(grpModuleSending, SWT.BORDER);
         ref_temp.setBounds(133, 50, 75, 27);
         
         btnStart = new Button(grpModuleSending, SWT.NONE);
         btnStart.setBounds(134, 99, 88, 29);
         btnStart.setText("Start");
         
         btnStop = new Button(grpModuleSending, SWT.NONE);
         btnStop.setBounds(228, 99, 88, 29);
         btnStop.setText("Stop");
         
         Label lblStatus = new Label(shlModule, SWT.NONE);
         lblStatus.setBounds(10, 216, 50, 17);
         lblStatus.setText("Status: ");    }
 }

SendingThread.java

package MainPackage;
import javax.jms.Connection;
 import javax.jms.DeliveryMode;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class SendingThread extends M2MainScreen  implements Runnable {
    public  Destination dest_ref_temp;
     public  MessageProducer producer_ref_temp;
     public  String user=null,password=null,host=null,text = null;
     
     public int port;
     public Session session_ref_temp;
     
     @Override
     public void run() {
         //while (isRunning) {
         // TODO Auto-generated method stub
         String user = env("ACTIVEMQ_USER", "admin");
         String password = env("ACTIVEMQ_PASSWORD", "password");
         String host = env("ACTIVEMQ_HOST", "localhost");
         int port = Integer.parseInt(env("ACTIVEMQ_PORT", "61616"));
         
         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://" + host + ":" + port);
        Connection connection = null;
         try {
             connection = factory.createConnection(user, password);
             connection.start();
             session_ref_temp = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                             
             dest_ref_temp = session_ref_temp.createQueue("REF_TEMP.FOO");
                   
             producer_ref_temp = session_ref_temp.createProducer(dest_ref_temp);
            
             producer_ref_temp.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
             
         } catch (JMSException e1) {
             // TODO Auto-generated catch block
             e1.printStackTrace();
         }
         
         //Destination dest = new ActiveMQTopic(destination);
        
        // Create a messages
         
         while(true){
     
             String text_ref_temp = string_ref_temp;
       
             if(isRunning=="false"){ 
                 i=0;
                 break;
              
             } 
             
        
             text_ref_temp=Integer.toString(i);
             i++;
             TextMessage message_ref_temp;
             
             
             try {
                 message_ref_temp = session_ref_temp.createTextMessage(text_ref_temp);
                      
                 producer_ref_temp.send(message_ref_temp);

                 //System.out.println("Send1: " + text);
                 doUpdate(display,ref_temp,text_ref_temp);
                
                 
                 try {
                       Thread.sleep(500);
                     } catch(InterruptedException e) {
                     }
             } catch (JMSException e1) {
                 // TODO Auto-generated catch block
                 e1.printStackTrace();
             } 
             
             
         
         
         //producer.send(session.createTextMessage("SHUTDOWN"));
         /*try {
             connection.close();
         } catch (JMSException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }*/
         }
         
             
     }
     private static String env(String key, String defaultValue) {
         String rc = System.getenv(key);
         if( rc== null )
             return defaultValue;
         return rc;
     }
}