ReplyTo in ZeroMQ using WS-Addressing

build.sh

mvn package

Callback.java

package org.ossandme.client;

import javax.jws.WebParam;
import javax.jws.WebService;
import javax.xml.ws.RequestWrapper;

@WebService
public class Callback {

    @RequestWrapper(localName = "sayHiResponse", targetNamespace = "http://apache.org/hello_world_soap_zmq/types")
    public void callback(@WebParam(targetNamespace = "http://apache.org/hello_world_soap_zmq/types", name = "responseType") String callbackMessage) {
        System.out.println("ASYNC REPLY RECEIVED FROM SERVER: " + callbackMessage);
        System.exit(0);
    }
}

Client.java

package org.ossandme.client;

...

public class Client {

    public static void doOperation() throws Exception {

        // Start callback service
        Endpoint.publish("zmq:(tcp://127.0.0.1:5555?socketOperation=bind&socketType=pull)", new Callback());

        // Create client
        SOAPService service = new SOAPService(Client.class.getResource("/hello_world_addr.wsdl").toURI().toURL());
        Greeter port = service.getSoapPort(new WebServiceFeature[]{new AddressingFeature()});

        // Add ReplyTo property to request
        Map<String, Object> requestContext = ((BindingProvider) port).getRequestContext();
        requestContext.put(CLIENT_ADDRESSING_PROPERTIES, createMaps());

        // Dispatch request
        port.sayHiAsync();
        System.out.println("CLIENT DISPATCHED REQUEST");
    }

    private static AddressingProperties createMaps() {
        AddressingProperties maps = AddressingBuilder.getAddressingBuilder().newAddressingProperties();
        EndpointReferenceType ert = new EndpointReferenceType();

        AttributedURIType replyTo = new AttributedURIType();
        replyTo.setValue("zmq:(tcp://127.0.0.1:5555?socketOperation=connect&socketType=push)");
        ert.setAddress(replyTo);

        maps.setReplyTo(ert);

        return maps;
    }

}

generate-client.sh

./wsdl2java -frontend jaxws21 -asyncMethods -client hello_world_addr.wsdl

generate-server.sh

./wsdl2java -frontend jaxws21 -server hello_world_addr.wsdl

gistfile1.txt

Jan 27, 2013 12:24:34 PM org.apache.cxf.service.factory.ReflectionServiceFactoryBean buildServiceFromClass
INFO: Creating Service {http://server.ossandme.org/}GreeterImplService from class org.ossandme.server.Greeter
Jan 27, 2013 12:24:34 PM org.apache.cxf.endpoint.ServerImpl initDestination
INFO: Setting the server's publish address to be zmq:(tcp://127.0.0.1:9000?socketOperation=bind&socketType=pull)
SERVER STARTED!!
Jan 27, 2013 12:24:34 PM org.apache.cxf.service.factory.ReflectionServiceFactoryBean buildServiceFromClass
INFO: Creating Service {http://client.ossandme.org/}CallbackService from class org.ossandme.client.Callback
Jan 27, 2013 12:24:34 PM org.apache.cxf.endpoint.ServerImpl initDestination
INFO: Setting the server's publish address to be zmq:(tcp://127.0.0.1:5555?socketOperation=bind&socketType=pull)
Jan 27, 2013 12:24:34 PM org.apache.hello_world_soap_zmq.SOAPService <clinit>
INFO: Can not initialize the default wsdl from hello_world_addr.wsdl
Jan 27, 2013 12:24:35 PM org.apache.cxf.service.factory.ReflectionServiceFactoryBean buildServiceFromWSDL
INFO: Creating Service {http://apache.org/hello_world_soap_zmq}SOAPService from WSDL: jar:file:/Users/claudemamo/playground/zmq-replyto-soap/target/zmq-replyto-soap-1.0-SNAPSHOT.jar!/hello_world_addr.wsdl
CLIENT DISPATCHED REQUEST
SERVER EXECUTING OPERATION sayHi...
ASYNC REPLY RECEIVED FROM SERVER: Bonjour

GreeterImpl.java

package org.ossandme.server;

import javax.xml.ws.soap.Addressing;

@Addressing
public class GreeterImpl implements Greeter {

    public String sayHi() {
        System.out.println("SERVER EXECUTING OPERATION sayHi...");
        return "Bonjour";
    }

}

hello_world_addr.wsdl

<wsdl:definitions name="HelloWorld" targetNamespace="http://apache.org/hello_world_soap_zmq"
                  xmlns="http://schemas.xmlsoap.org/wsdl/"
                  xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
                  xmlns:tns="http://apache.org/hello_world_soap_zmq"
                  xmlns:x1="http://apache.org/hello_world_soap_zmq/types"
                  xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
                  xmlns:zmq="http://cxf.apache.org/transports/zmq"
                  xmlns:xsd="http://www.w3.org/2001/XMLSchema">

    <wsdl:types>
        ...
    </wsdl:types>

    <wsdl:message name="sayHiRequest">
        <wsdl:part element="x1:sayHi" name="in"/>
    </wsdl:message>
    <wsdl:message name="sayHiResponse">
        <wsdl:part element="x1:sayHiResponse" name="out"/>
    </wsdl:message>

    <wsdl:portType name="Greeter">
        <wsdl:operation name="sayHi">
            <wsdl:input message="tns:sayHiRequest" name="sayHiRequest"/>
            <wsdl:output message="tns:sayHiResponse" name="sayHiResponse"/>
        </wsdl:operation>
    </wsdl:portType>

    <wsdl:binding name="Greeter_SOAPBinding" type="tns:Greeter">
        <soap:binding style="document" transport="http://cxf.apache.org/transports/zmq"/>
        <wsdl:operation name="sayHi">
            <soap:operation soapAction="" style="document"/>
            <wsdl:input name="sayHiRequest">
                <soap:body use="literal"/>
            </wsdl:input>
            <wsdl:output name="sayHiResponse">
                <soap:body use="literal"/>
            </wsdl:output>
        </wsdl:operation>
    </wsdl:binding>

    <wsdl:service name="SOAPService">
        <wsdl:port binding="tns:Greeter_SOAPBinding" name="SoapPort">
            <zmq:address location="tcp://127.0.0.1:9000"/>
            <zmq:serviceConfig socketOperation="bind" socketType="pull"/>
            <zmq:clientConfig socketOperation="connect" socketType="push"/>
        </wsdl:port>
    </wsdl:service>

</wsdl:definitions>

Main.java

package org.ossandme;

import org.ossandme.client.Client;
import org.ossandme.server.Server;

public class Main {

    public static void main(String args[]) throws Exception {
        Server.start();
        Client.doOperation();

        // Wait for callback to receive async reply
        Thread.sleep(5 * 60 * 1000);
    }

}

run.sh

java -Djava.library.path=[jzmq.path] -cp target/zmq-replyto-soap-1.0-SNAPSHOT.jar:target/lib/* org.ossandme.Main

Server.java

package org.ossandme.server;

import javax.xml.ws.Endpoint;

public class Server {

    public static void start() throws Exception {
        Endpoint.publish("zmq:(tcp://127.0.0.1:9000?socketOperation=bind&socketType=pull)", new GreeterImpl());
        System.out.println("SERVER STARTED!!");
    }
}