org.zeromq

Class ZProxy

java.lang.Object

org.zeromq.ZProxy

public class ZProxy
extends Object

Implementation of a remotely controlled proxy for 0MQ, using ZActor.
The goals of this implementation are to delegate the creation of sockets in a background thread via a callback interface to ensure their correct use and to provide ultimately to end-users the following features.

Basic features:

• Remote Control
  • Start: if was paused, flushes the pending messages
  • Pause: lets the socket queues accumulate messages according to their types
  • Stop: Shutdowns the proxy, can be restarted
  • Status: Retrieves the status of the proxy
  • Cold Restart: Closes and recreates the connections
  • Hot Restart: User-defined behavior with custom messages
  • Configure: User-defined behavior with custom messages
  • ...: Custom commands of your own
  • Exit: Definitive shutdown of the proxy and its control
  All the non-custom commands can be performed in asynchronous or synchronous mode.
  
  
• Proxy mechanism ensured by pluggable pumps
  • with built-in low-level ZmqPump (zmq.ZMQ): useful for performances
  • with built-in high-level ZProxy.ZPump (ZeroMQ): useful for message transformation, lower performances
  • with your own-custom proxy pump implementing a 1-method interface

You can have all the above non-customizable features in about these lines of code:

 final ZProxy.Proxy provider = new ZProxy.SimpleProxy()
        {
            public Socket create(ZContext ctx, ZProxy.Plug place, Object[] args)
            {
                assert ("TEST".equals(args[0]);
                Socket socket = null;
                if (place == ZProxy.Plug.FRONT) {
                    socket = ctx.createSocket(ZMQ.ROUTER);
                }
                if (place == ZProxy.Plug.BACK) {
                    socket = ctx.createSocket(ZMQ.DEALER);
                }
                return socket;
            }

            public void configure(Socket socket, ZProxy.Plug place, Object[] args)
            {
                assert ("TEST".equals(args[0]);
                int port = -1;
                if (place == ZProxy.Plug.FRONT) {
                    port = socket.bind("tcp://127.0.0.1:6660");
                }
                if (place == ZProxy.Plug.BACK) {
                    port = socket.bind("tcp://127.0.0.1:6661");
                }
                if (place == ZProxy.Plug.CAPTURE && socket != null) {
                    socket.bind("tcp://127.0.0.1:4263");
                }
            }
        };

        ZProxy proxy = ZProxy.newProxy("ProxyOne", provider, "ABRACADABRA", Arrays.asList("TEST"));

 

Once created, the proxy is not started. You have to perform first a start command on it. This choice was made because it is easier for a user to start it with one line of code than for the code to internally handle different possible starting states (after all, someone may want the proxy started but paused at first or configured in a specific way?) and because the a/sync stuff was funnier. Life is unfair ... Or maybe an idea is floating in the air?
You can then use it like this:

 final boolean async = false, sync = true;
        String status = null;
        status = proxy.status();
        status = proxy.pause(sync);
        status = proxy.start(async);
        ZMsg msg = proxy.restart(new ZMsg());
        status = proxy.status(async);
        status = proxy.stop(sync);
        boolean here = proxy.sign();
        ZMsg cfg = new ZMsg();
        msg.add("CONFIG-1");
        ZMsg rcvd = proxy.configure(cfg);
        proxy.exit(async);
        status = proxy.status(sync);
        assert (!proxy.started());
   
 

A programmatic interface with enums is also available.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	ZProxy.Command
static class	ZProxy.Plug Possible places for sockets in the proxy.
static interface	ZProxy.Proxy
static interface	ZProxy.Pump
static class	ZProxy.State
static class	ZProxy.ZPump A pump that reads a message as a whole before transmitting it.

Field Summary

Fields

Modifier and Type	Field and Description
static String	ALIVE
static String	EXITED
static String	PAUSED
static String	STARTED
static String	STOPPED

Constructor Summary

Constructors

Constructor and Description
ZProxy(String name, ZAgent.SelectorCreator selector, ZProxy.Proxy creator, String motdelafin, Object... args) Creates a new named proxy.
ZProxy(ZAgent.SelectorCreator selector, ZProxy.Proxy creator, String motdelafin, Object... args) Creates a new unnamed proxy.
ZProxy(ZContext ctx, String name, ZAgent.SelectorCreator selector, ZProxy.Proxy sockets, ZProxy.Pump pump, String motdelafin, Object... args) Creates a new named proxy.

Method Summary

Methods

Modifier and Type	Method and Description
String	command(String command, boolean sync) Sends a command message to the proxy actor.
ZProxy.State	command(ZProxy.Command command, boolean sync) Sends a command message to the proxy actor.
ZMsg	command(ZProxy.Command command, ZMsg msg, boolean sync) Sends a command message to the proxy actor.
ZMsg	configure(ZMsg msg) Configures the proxy.
String	exit(boolean sync) Stops the proxy and exits.
boolean	isStarted() Binary inquiry for the status of the proxy.
static ZProxy	newProxy(ZContext ctx, String name, ZAgent.SelectorCreator selector, ZProxy.Proxy sockets, String motdelafin, Object... args) Creates a new low-level proxy for better performances.
static ZProxy	newProxy(ZContext ctx, String name, ZProxy.Proxy sockets, String motdelafin, Object... args)
static ZProxy	newZProxy(ZContext ctx, String name, ZAgent.SelectorCreator selector, ZProxy.Proxy sockets, String motdelafin, Object... args) Creates a new proxy in a ZeroMQ way.
static ZProxy	newZProxy(ZContext ctx, String name, ZProxy.Proxy sockets, String motdelafin, Object... args)
String	pause(boolean sync) Pauses the proxy.
String	restart(ZMsg hot) Restarts the proxy.
String	start(boolean sync) Starts the proxy.
boolean	started() Binary inquiry for the status of the proxy.
String	status() Inquires for the status of the proxy.
String	status(boolean sync) Inquires for the status of the proxy.
String	stop(boolean sync) Stops the proxy.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

STARTED

public static final String STARTED

PAUSED

public static final String PAUSED

STOPPED

public static final String STOPPED

EXITED

public static final String EXITED

ALIVE

public static final String ALIVE

Constructor Detail

ZProxy

public ZProxy(ZAgent.SelectorCreator selector,
      ZProxy.Proxy creator,
      String motdelafin,
      Object... args)

Creates a new unnamed proxy.

Parameters:

selector - the creator of the selector used for the proxy.

creator - the creator of the sockets of the proxy.

ZProxy

public ZProxy(String name,
      ZAgent.SelectorCreator selector,
      ZProxy.Proxy creator,
      String motdelafin,
      Object... args)

Creates a new named proxy.

Parameters:

name - the name of the proxy (used in threads naming).

selector - the creator of the selector used for the proxy.

creator - the creator of the sockets of the proxy.

ZProxy

public ZProxy(ZContext ctx,
      String name,
      ZAgent.SelectorCreator selector,
      ZProxy.Proxy sockets,
      ZProxy.Pump pump,
      String motdelafin,
      Object... args)

Creates a new named proxy.

Parameters:

ctx - the main context used. If null, a new context will be created and closed at the stop of the operation. If not null, it is the responsibility of the call to close it.

name - the name of the proxy (used in threads naming).

selector - the creator of the selector used for the proxy.

sockets - the creator of the sockets of the proxy.

pump - the pump used for the proxy

Method Detail

newZProxy

public static ZProxy newZProxy(ZContext ctx,
               String name,
               ZAgent.SelectorCreator selector,
               ZProxy.Proxy sockets,
               String motdelafin,
               Object... args)

Creates a new proxy in a ZeroMQ way. This proxy will be less efficient than the low-level one.

Parameters:

ctx - the context used for the proxy. Possibly null, in this case a new context will be created and automatically destroyed afterwards.

name - the name of the proxy. Possibly null.

selector - the creator of the selector used for the internal polling. Not null.

sockets - the sockets creator of the proxy. Not null.

args - an optional array of arguments that will be passed at the creation.

Returns:

the created proxy.

newZProxy

public static ZProxy newZProxy(ZContext ctx,
               String name,
               ZProxy.Proxy sockets,
               String motdelafin,
               Object... args)

newProxy

public static ZProxy newProxy(ZContext ctx,
              String name,
              ZAgent.SelectorCreator selector,
              ZProxy.Proxy sockets,
              String motdelafin,
              Object... args)

Creates a new low-level proxy for better performances.

Parameters:

ctx - the context used for the proxy. Possibly null, in this case a new context will be created and automatically destroyed afterwards.

name - the name of the proxy. Possibly null.

selector - the creator of the selector used for the internal polling. Not null.

sockets - the sockets creator of the proxy. Not null.

args - an optional array of arguments that will be passed at the creation.

Returns:

the created proxy.

newProxy

public static ZProxy newProxy(ZContext ctx,
              String name,
              ZProxy.Proxy sockets,
              String motdelafin,
              Object... args)

start

public String start(boolean sync)

Starts the proxy.

Parameters:

sync - true to read the status in synchronous way, false for asynchronous mode

Returns:

the read status

pause

public String pause(boolean sync)

Pauses the proxy. A paused proxy will cease processing messages, causing them to be queued up and potentially hit the high-water mark on the frontend or backend socket, causing messages to be dropped, or writing applications to block.

Parameters:

sync - true to read the status in synchronous way, false for asynchronous mode

Returns:

the read status

stop

public String stop(boolean sync)

Stops the proxy.

Parameters:

sync - true to read the status in synchronous way, false for asynchronous mode

Returns:

the read status

command

public String command(String command,
             boolean sync)

Sends a command message to the proxy actor. Can be useful for programmatic interfaces. Does not works with commands CONFIG and RESTART.

Parameters:

command - the command to execute.

sync - true to read the status in synchronous way, false for asynchronous mode

Returns:

the read status

command

public ZProxy.State command(ZProxy.Command command,
                   boolean sync)

Sends a command message to the proxy actor. Can be useful for programmatic interfaces. Does not works with commands CONFIG and RESTART.

Parameters:

command - the command to execute.

sync - true to read the status in synchronous way, false for asynchronous mode

Returns:

the read state

command

public ZMsg command(ZProxy.Command command,
           ZMsg msg,
           boolean sync)

Sends a command message to the proxy actor. Can be useful for programmatic interfaces. Works only with commands CONFIG and RESTART.

Parameters:

command - the command to execute.

msg - the custom message to transmit.

sync - true to read the status in synchronous way, false for asynchronous mode

Returns:

the response message

configure

public ZMsg configure(ZMsg msg)

Configures the proxy. The distant side has to send back one (1) mandatory response message.

Parameters:

msg - the custom message sent as configuration tip

Returns:

the mandatory response message of the configuration.

restart

public String restart(ZMsg hot)

Restarts the proxy. Stays alive.

Parameters:

hot - null to make a cold restart (closing then re-creation of the sockets) or a configuration message to perform a configurable hot restart,

exit

public String exit(boolean sync)

Stops the proxy and exits.

Parameters:

sync - true to read the status in synchronous way, false for asynchronous mode

Returns:

the read status

status

public String status()

Inquires for the status of the proxy. This call is synchronous.

status

public String status(boolean sync)

Inquires for the status of the proxy.

Parameters:

sync - true to read the status in synchronous way, false for asynchronous mode. If false, you get the last cached status of the proxy

isStarted

public boolean isStarted()

Binary inquiry for the status of the proxy.

started

public boolean started()

Binary inquiry for the status of the proxy.