I am trying to implement using XPUB and XSUB as shown in the picture below. I looked at their examples, but could not get them for XPUB and XSUB in Java. Here they gave an example in C, which is a bit complicated since I am new to ZeroMQ.
I am trying to use it in android using jni wrapped version . Please help me find an example of how to implement this Pub-Sub network with a proxy server in ZeroMQ using java.
I am currently linking to http://zguide.zeromq.org/page:all
I tried to transfer it as follows. Subscriber.java
public class Subscriber extends Thread implements Runnable {
private static final String TAG = "Subscriber"; private Context ctx; public Subscriber(ZMQ.Context z_context) { this.ctx = z_context; } @Override public void run() { super.run(); ZMQ.Socket mulServiceSubscriber = ctx.socket(ZMQ.SUB); mulServiceSubscriber.connect("tcp://localhost:6001"); mulServiceSubscriber.subscribe("A".getBytes()); mulServiceSubscriber.subscribe("B".getBytes()); while (true) { Log.d(TAG, "Subscriber loop started.."); String content = new String(mulServiceSubscriber.recv(0)); Log.d(TAG, "Subscriber Received : "+content); } }
}
Publisher.java
public class Publisher extends Thread implements Runnable {
private static final String TAG = "Publisher"; private Context ctx; public Publisher(ZMQ.Context z_context) { this.ctx = z_context; } @Override public void run() { super.run(); ZMQ.Socket publisher = ctx.socket(ZMQ.PUB); publisher.connect("tcp://localhost:6000"); while (true) { Log.d(TAG, "Publisher loop started.."); publisher.send(("A Hello " + new Random(100).nextInt()).getBytes() , 0); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } }
}
XListener.java (currently a simple forwarder)
Public class XListener extends Thread implements Runnable {
private static final String TAG = null; private Socket publisherX; private Context ctx; private Socket subscriberX; public XListener(ZMQ.Context ctx, ZMQ.Socket subscriberX, ZMQ.Socket publisherX) { this.ctx = ctx; this.subscriberX = subscriberX; this.publisherX = publisherX; } @Override public void run() { super.run(); while (true) { Log.d(TAG, "XListener loop started.."); String msg = new String(subscriberX.recvStr()); Log.v(TAG, "Listener Received: " +"MSG :"+msg); publisherX.send(msg.getBytes(), 0); } }
}
in main () application
private void main () {ZMQ.Context ctx = ZMQ.context (1);
ZMQ.Socket subscriberX = ctx.socket(ZMQ.XSUB); subscriberX.bind("tcp://*:6000"); try { Thread.sleep(500); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } ZMQ.Socket publisherX = ctx.socket(ZMQ.XPUB); publisherX.bind("tcp://*:6001"); try { Thread.sleep(500); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } new XListener(ctx, subscriberX, publisherX).start(); try { Thread.sleep(500); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } new XSender(ctx, subscriberX, publisherX).start(); try { Thread.sleep(500); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } new Subscriber(ctx).start(); new Publisher(ctx).start(); } With code, I can not listen to XSUB . When porting espresso.c, I could not find any shell in the ZMQ java bindings. How to implement a simple proxy or am I missing something?