Zero MQ Pub Sub Simple Sample

The following is a simple 0MQ tcp pub sub simple sample:
The main:

        ZMQ.Context context = ZMQ.context(1);
   
        ZeroMqPubServer theZeroMqPubServer = new ZeroMqPubServer(context);
        theZeroMqPubServer.start();
      
        ZeroMqSubClient theZeroMqSubClient = new ZeroMqSubClient(context);
        
        theZeroMqSubClient.start();

The server

import java.util.logging.Level;
import java.util.logging.Logger;
import org.zeromq.ZMQ;
public class ZeroMqPubServer  extends Thread{
    
    ZMQ.Context  mContext;
    public ZeroMqPubServer(ZMQ.Context pContext) {
        this.mContext = pContext;
    }
    
     @Override
     public void run()
     {
         ZMQ.Socket publisher = mContext.socket(ZMQ.PUB);
        publisher.bind("tcp://*:5556");
        int theCounter = 0 ; 
         while (!Thread.currentThread ().isInterrupted ()) {
            try {
                theCounter++;
                //  Send message to all subscribers
                String update = String.format("MyTopic %d ", theCounter);
                
                publisher.send(update, 0);
                
                Thread.sleep(100);
            }
            // context.term ();
            catch (InterruptedException ex) {
                Logger.getLogger(ZeroMqPubServer.class.getName()).log(Level.SEVERE, null, ex);
            }          
        }
     }
}

The client

import java.nio.charset.Charset;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.zeromq.ZMQ;
public class ZeroMqSubClient  extends Thread{
    
      ZMQ.Context  mContext;
    public ZeroMqSubClient(ZMQ.Context pContext) {
        this.mContext = pContext;
    }
    
    
     @Override
     public void run()
     {
          try {
              Thread.sleep(3000 );
              
              ZMQ.Socket subscriber = mContext.socket(ZMQ.SUB);
              subscriber.connect("tcp://localhost:5556");
              
              //  Subscribe to zipcode, default is NYC, 10001
              String filter =  "MyTopic";
              subscriber.subscribe(filter.getBytes());
              
              while (!Thread.currentThread ().isInterrupted ()) {
                  
                  //  Use trim to remove the tailing '0' character
                  String theStr = subscriber.recvStr(Charset.defaultCharset());
                  
                  System.out.println("Recieved:" + theStr);
              }
              subscriber.close();
              //  mContext.term();
          } catch (InterruptedException ex) {
              Logger.getLogger(ZeroMqSubClient.class.getName()).log(Level.SEVERE, null, ex);
          }
    }
}

The result
debug:
Start
Recieved:MyTopic 31
Recieved:MyTopic 32
Recieved:MyTopic 33
Recieved:MyTopic 34
Recieved:MyTopic 35
Recieved:MyTopic 36
Recieved:MyTopic 37
Recieved:MyTopic 38
Recieved:MyTopic 39
Recieved:MyTopic 40
Recieved:MyTopic 41
Recieved:MyTopic 42
Recieved:MyTopic 43
Recieved:MyTopic 44
Recieved:MyTopic 45
Recieved:MyTopic 46
Recieved:MyTopic 47
Recieved:MyTopic 48
Recieved:MyTopic 49
Recieved:MyTopic 50
Recieved:MyTopic 51
Recieved:MyTopic 52
Recieved:MyTopic 53
Recieved:MyTopic 54
Recieved:MyTopic 55

How to run ZeroMQ with Java in windows7

In order to run 0MQ with java in windows follow the following steps:
1.Compile the libzmq dll 


Set the output lib of the libzmq project

Don’t forget to compile for x64 Bit

The compilation result should be:
1> Creating library libzmq.lib and object libzmq.exp

 

1> libzmq.vcxproj -> H:\deleteme100\zeromq-4.0.4\builds\msvc\libzmq\../../../bin/x64/libzmq_d.dll

========== Rebuild All: 1 succeeded, 0 failed, 0 skipped ==========

 

2.Compile the jzmq dll.

Set the libzmq library in the project additional dependencies

Set the jdk and the zero mq header files pathes in the c++ include directory .

Compile the project in x64 bit .
The output should be :

1>------ Rebuild All started: Project: jzmq, Configuration: Release x64 ------

1> The syntax of the command is incorrect.

1> Note: Some input files use or override a deprecated API.

1> Note: Recompile with -Xlint:deprecation for details.

1> Generating JNI header

1> Generating JNI header

1> Generating JNI header

1> Generating JNI header

1> Generating JNI header

1> Generating JNI header

1> Generating JNI header

1> Context.cpp

1> Poller.cpp

1>..\..\..\src\main\c++\Poller.cpp(76): warning C4244: '=' : conversion from 'jint' to 'short', possible loss of data

1> Socket.cpp

1>..\..\..\src\main\c++\Socket.cpp(266): warning C4267: 'argument' : conversion from 'size_t' to 'jsize', possible loss of data

1>..\..\..\src\main\c++\Socket.cpp(272): warning C4267: 'argument' : conversion from 'size_t' to 'jsize', possible loss of data

1>..\..\..\src\main\c++\Socket.cpp(834): warning C4267: 'initializing' : conversion from 'size_t' to 'int', possible loss of data

1>..\..\..\src\main\c++\Socket.cpp(862): warning C4267: 'initializing' : conversion from 'size_t' to 'int', possible loss of data

1> util.cpp

1> ZMQ.cpp

1>C:\Program Files (x86)\MSBuild\Microsoft.Cpp\v4.0\V110\Microsoft.CppBuild.targets(1137,5): warning MSB8012: TargetPath(H:\deleteme101\jzmq-master\builds\msvc\x64\Release\jzmq.dll) does not match the Linker's OutputFile property value (H:\deleteme101\jzmq-master\lib\jzmq.dll). This may cause your project to build incorrectly. To correct this, please make sure that $(OutDir), $(TargetName) and $(TargetExt) property values match the value specified in %(Link.OutputFile).

1> Creating library H:\deleteme101\jzmq-master\builds\msvc\x64\Release\jzmq.lib and object H:\deleteme101\jzmq-master\builds\msvc\x64\Release\jzmq.exp

1> Generating code

1> Finished generating code

1> jzmq.vcxproj -> H:\deleteme101\jzmq-master\builds\msvc\x64\Release\jzmq.dll

1> added manifest

1> adding: org/zeromq/App.class(in = 2433) (out= 1352)(deflated 44%)

1> adding: org/zeromq/EmbeddedLibraryTools.class(in = 4442) (out= 2473)(deflated 44%)

1> adding: org/zeromq/ZContext.class(in = 2621) (out= 1333)(deflated 49%)

1> adding: org/zeromq/ZDispatcher$1.class(in = 206) (out= 161)(deflated 21%)

1> adding: org/zeromq/ZDispatcher$SocketDispatcher$1.class(in = 861) (out= 413)(deflated 52%)

1> adding: org/zeromq/ZDispatcher$SocketDispatcher$2.class(in = 2034) (out= 829)(deflated 59%)

1> adding: org/zeromq/ZDispatcher$SocketDispatcher$ZMessageBuffer.class(in = 1395) (out= 653)(deflated 53%)

1> adding: org/zeromq/ZDispatcher$SocketDispatcher.class(in = 4184) (out= 1790)(deflated 57%)

1> adding: org/zeromq/ZDispatcher$ZMessageHandler.class(in = 340) (out= 213)(deflated 37%)

1> adding: org/zeromq/ZDispatcher$ZSender.class(in = 780) (out= 436)(deflated 44%)

1> adding: org/zeromq/ZDispatcher.class(in = 2778) (out= 1254)(deflated 54%)

1> adding: org/zeromq/ZFrame.class(in = 4765) (out= 2642)(deflated 44%)

1> adding: org/zeromq/ZMQ$Context.class(in = 1262) (out= 711)(deflated 43%)

1> adding: org/zeromq/ZMQ$Error.class(in = 2545) (out= 1433)(deflated 43%)

1> adding: org/zeromq/ZMQ$Event.class(in = 1444) (out= 732)(deflated 49%)

1> adding: org/zeromq/ZMQ$Poller.class(in = 4092) (out= 2007)(deflated 50%)

1> adding: org/zeromq/ZMQ$PollItem.class(in = 1651) (out= 736)(deflated 55%)

1> adding: org/zeromq/ZMQ$Socket.class(in = 11809) (out= 4589)(deflated 61%)

1> adding: org/zeromq/ZMQ.class(in = 3529) (out= 1717)(deflated 51%)

1> adding: org/zeromq/ZMQException.class(in = 771) (out= 485)(deflated 37%)

1> adding: org/zeromq/ZMQForwarder.class(in = 1464) (out= 870)(deflated 40%)

1> adding: org/zeromq/ZMQQueue.class(in = 1860) (out= 1041)(deflated 44%)

1> adding: org/zeromq/ZMQStreamer.class(in = 424) (out= 260)(deflated 38%)

1> adding: org/zeromq/ZMsg.class(in = 10214) (out= 4286)(deflated 58%)

========== Rebuild All: 1 succeeded, 0 failed, 0 skipped ==========

3.Copy the generated dlls into the system 32 directory

Note:
Copy the libzmq_d.dll into libzmq.dll

4.Add the zmq jar to your project
found in zmq-master\lib
 

5.Enjoy 0mq.

References:
http://stackoverflow.com/questions/11455588/how-to-run-zeromq-with-java

Simple 0MQ request replay client and server

The biggest problem of 0MQ in my opinion is it’s installation process.
In contradiction to Rabbit , MSMQ and ActiveMQ its very hard to install it and use it.
But there are a lot of samples in any language.
The project:

The main

package test0mq2;
import java.lang.reflect.Field;
public class Test0Mq2 {
    public static void main(String[] args) throws Exception {
 
        SetLibraryPath();
 
        StartServer();
        
        SendMsgToServer();
       
    }
    private static void StartServer()
    {
        ZeroMQRepServer theZeroMQRepServer = new ZeroMQRepServer();
    
        theZeroMQRepServer.Startlisten();
    }
    
    private static void SendMsgToServer()
    {
        ZeroMQREQClient theZeroMQREQClient = new ZeroMQREQClient();
        
        theZeroMQREQClient.SendMsg ("Hello Zvika");
    }
    private static void SetLibraryPath() throws SecurityException, IllegalArgumentException, IllegalAccessException, NoSuchFieldException {
        String libPathProperty = System.getProperty("java.library.path");
        
        System.out.println(libPathProperty);
        
        System.setProperty( "java.library.path", "H:\\deleteme100\\zeromq-4.0.4\\bin\\Win32" );
        
        Field fieldSysPath = ClassLoader.class.getDeclaredField( "sys_paths" );
        fieldSysPath.setAccessible( true );
        fieldSysPath.set( null, null );
    
        libPathProperty = System.getProperty("java.library.path");
        System.out.println(libPathProperty);
    }
}

Note:
The SetLibraryPath method is used to set the path of the jzmq
The jzmq.dll is a jni native extension used for the 0MQ
The Server

package test0mq2;
import org.zeromq.ZMQ;
public class ZeroMQRepServer extends Thread{
    
     public void Startlisten ()
     {
         (new ZeroMQRepServer()).start();
     }
     @Override
     public void run() {
          
        ZMQ.Context context = ZMQ.context(1);
        //  Socket to talk to clients
        ZMQ.Socket responder = context.socket(ZMQ.REP);
        responder.bind("tcp://*:5555");
        while (!Thread.currentThread().isInterrupted()) {
            // Wait for next request from the client
            byte[] request = responder.recv(0);
           
            // Send reply back to client
            String theReplay = "Get:" + new String(request);
            responder.send(theReplay.getBytes(), 0);
        }
        responder.close();
        context.term();
     }
}

 The Client

package test0mq2;
import org.zeromq.ZMQ;
public class ZeroMQREQClient {
    
    public void SendMsg (String pMsg)
    {
        ZMQ.Context context = ZMQ.context(1);
       //Connecting to server with REQ mode 
        ZMQ.Socket requester = context.socket(ZMQ.REQ);
        
        requester.connect("tcp://localhost:5555");
        requester.send(pMsg.getBytes(), 0);
        byte[] reply = requester.recv(0);
        
        System.out.println("Received:" + new String(reply));
        
        requester.close();
        
        context.term();
    }
}