Request / Response and the SendRequest Bean

Blocking request

MUX mux = (MUX) NameRegistrar.get("mux.upstream");
ISOMsg resp = mux.request(m, 30000);
if (resp != null) {
    // process response
} else {
    // timed out
}

request() blocks the calling thread until either a matching response arrives or the timeout elapses. The timeout is in milliseconds. A null return means no response was received in time; the remote host may be down, slow, or the STAN was never echoed back.

note

request() throws ISOException if the MUX is not connected or if a duplicate key is detected. Always check mux.isConnected() before sending, or handle the exception appropriately.

Async request

For non-blocking callers, the async variant accepts an ISOResponseListener:

mux.request(m, 30000, new ISOResponseListener() {
    @Override
    public void responseReceived(ISOMsg response, Object handBack) {
        // runs on a Space/scheduler thread
    }
    @Override
    public void expired(Object handBack) {
        // timeout
    }
}, myHandBackObject);

The handBack is an arbitrary object passed through unchanged to the listener — useful for carrying per-request context (transaction ID, callback references, etc.) without additional maps.

Internally, the async path stores an AsyncRequest in isp instead of the raw ISOMsg. When notify() finds an AsyncRequest, it calls responseReceived() directly rather than posting to isp for a blocking thread to collect.

isConnected()

if (!mux.isConnected()) {
    // channel is down, don't bother sending
}

isConnected() inspects the ready semaphores configured via <ready>. It returns true if at least one of them is present in Space — meaning the corresponding ChannelAdaptor has an active TCP connection. If <ready> is not configured, isConnected() returns true as long as the QMUX service itself is running (not recommended for production).

The SendRequest bean

The tutorial includes a SendRequest QBean that sends a 2800 Network Management Request every minute:

public class SendRequest extends QBeanSupport implements Runnable {
    private final AtomicInteger stan = new AtomicInteger(0);

    @Override
    protected void startService() {
        new Thread(this, "send-request").start();
    }

    @Override
    public void run() {
        while (running()) {
            try {
                MUX mux = (MUX) NameRegistrar.get("mux." + muxName);
                ISOMsg m = new ISOMsg();
                m.setMTI("2800");
                m.set(11, ISOUtil.zeropad(String.valueOf(stan.incrementAndGet()), 6));
                m.set(70, "301");  // Echo test

                getLog().info("sending 2800, STAN=" + m.getString(11));
                ISOMsg resp = mux.request(m, timeout);

                if (resp != null)
                    getLog().info("response 2810, STAN=" + resp.getString(11)
                        + " rc=" + resp.getString(39));
                else
                    getLog().warn("2800 timed out after " + timeout + "ms");

            } catch (NameRegistrar.NotFoundException e) {
                getLog().warn("MUX '" + muxName + "' not found, will retry");
            } catch (ISOException e) {
                getLog().warn(e);
            }
            ISOUtil.sleep(interval);
        }
    }
}

It is configured via src/dist/deploy/50_send_request.xml:

<send-request class="org.jpos.tutorial.SendRequest" logger="Q2" name="send-request">
  <property name="mux"      value="upstream" />
  <property name="interval" value="60000" />
  <property name="timeout"  value="30000" />
</send-request>

A few things worth noting about this implementation:

NameRegistrar.get("mux.upstream") is called inside the loop, not in initService(). This is intentional: QMUX registers itself in NameRegistrar during its own initService(), which runs at a deterministic deployment order. If SendRequest started at 50_ and QMUX at 20_, QMUX is always registered by the time SendRequest starts. But looking up inside the loop also means a restart of the QMUX bean doesn't require restarting SendRequest.

NameRegistrar.NotFoundException is caught and logged as a warning, then the loop sleeps and retries. This makes the bean resilient to deployment ordering surprises during development.

The STAN counter is an AtomicInteger. This works for a single SendRequest bean; in a multi-threaded production context the STAN source would be a shared, persistent counter (often stored in a database or Space).

ISOUtil.sleep(interval) at the end of the loop means the interval is wall-clock time between the end of one cycle and the start of the next, not a strict period. If the request() takes 500ms, the effective period is interval + 500ms. For a keep-alive probe this is fine.