Saggi Mizrahi has uploaded a new change for review.
Change subject: [WIP]Java Bindings: Proton support in Java Bindings
......................................................................
[WIP]Java Bindings: Proton support in Java Bindings
Change-Id: I94c52e118cb63d7df84b89a9b93da7b9e477be91
Signed-off-by: Saggi Mizrahi <smizrahi(a)redhat.com>
---
A
client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonAuthenticator.java
A client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonClient.java
A client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonListener.java
A client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonReactor.java
A
client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/AmqpReactorTestHelper.java
A
client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/TestJsonRpcClientAMQP.java
6 files changed, 844 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/28/15428/1
diff --git
a/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonAuthenticator.java
b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonAuthenticator.java
new file mode 100644
index 0000000..35c9099
--- /dev/null
+++
b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonAuthenticator.java
@@ -0,0 +1,98 @@
+package org.ovirt.vdsm.reactors;
+
+import java.io.IOException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.apache.qpid.proton.driver.Connector;
+import org.apache.qpid.proton.engine.Sasl;
+import org.apache.qpid.proton.engine.Sasl.SaslOutcome;
+import org.apache.qpid.proton.engine.Sasl.SaslState;
+import org.ovirt.vdsm.reactors.ProtonAuthenticator.AuthenticatorType;
+
+public final class ProtonAuthenticator {
+
+ public enum AuthenticatorType {
+
+ SERVER, CLIENT
+ }
+
+ public enum ConnectionState {
+
+ AUTHENTICATING, CONNECTED, FAILED
+ }
+ private ConnectionState _state;
+ final private AuthenticatorType _authType;
+ final private Connector<?> _connector;
+
+ public ProtonAuthenticator(Connector<?> connector,
+ AuthenticatorType authType) {
+ _authType = authType;
+ setState(ConnectionState.AUTHENTICATING);
+ _connector = connector;
+ final Sasl sasl = _connector.sasl();
+ if (authType == AuthenticatorType.CLIENT) {
+ sasl.setMechanisms(new String[]{"ANONYMOUS"});
+ sasl.client();
+ }
+ }
+
+ private void setState(ConnectionState state) {
+ _state = state;
+ }
+
+ public ConnectionState getState() {
+ return _state;
+ }
+
+ public void authenticate() {
+ final Sasl sasl = _connector.sasl();
+
+ while (true) {
+ try {
+ this._connector.process();
+ } catch (IOException ex) {
+ return;
+ }
+ final SaslState state = sasl.getState();
+ switch (state) {
+ case PN_SASL_CONF:
+ if (_authType == AuthenticatorType.SERVER) {
+ sasl.setMechanisms(new String[]{"ANONYMOUS"});
+ sasl.server();
+ }
+ break;
+ case PN_SASL_STEP:
+ if (_authType == AuthenticatorType.SERVER) {
+ final String[] mechs = sasl.getRemoteMechanisms();
+ if (mechs.length < 1) {
+ sasl.done(SaslOutcome.PN_SASL_AUTH);
+ break;
+ }
+
+ final String mech = mechs[0];
+ if (mech.equals("ANONYMOUS")) {
+ sasl.done(SaslOutcome.PN_SASL_OK);
+ } else {
+ sasl.done(SaslOutcome.PN_SASL_AUTH);
+ }
+ }
+ return;
+ case PN_SASL_PASS:
+ this.setState(ConnectionState.CONNECTED);
+ return;
+ case PN_SASL_FAIL:
+ this.setState(ConnectionState.FAILED);
+ return;
+ case PN_SASL_IDLE:
+
+ break;
+ default:
+ return;
+ }
+ }
+ }
+
+ public AuthenticatorType getAuthType() {
+ return _authType;
+ }
+}
diff --git
a/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonClient.java
b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonClient.java
new file mode 100644
index 0000000..4baffbf
--- /dev/null
+++ b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonClient.java
@@ -0,0 +1,224 @@
+package org.ovirt.vdsm.reactors;
+
+import java.nio.ByteBuffer;
+import java.util.Calendar;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Future;
+
+import javax.swing.event.EventListenerList;
+
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.messaging.Data;
+import org.apache.qpid.proton.amqp.messaging.Section;
+import org.apache.qpid.proton.engine.Delivery;
+import org.apache.qpid.proton.engine.EndpointState;
+import org.apache.qpid.proton.engine.Link;
+import org.apache.qpid.proton.engine.Receiver;
+import org.apache.qpid.proton.engine.Sender;
+import org.apache.qpid.proton.engine.Session;
+import org.apache.qpid.proton.message.Message;
+import org.apache.qpid.proton.message.MessageFactory;
+import org.apache.qpid.proton.message.impl.MessageFactoryImpl;
+
+public final class ProtonClient implements ReactorClient {
+ private final ProtonReactor _reactor;
+ final private Session _ssn;
+
+ private Sender _sender;
+ private Receiver _receiver;
+
+ private final int _CREDIT = 10;
+ private final ConcurrentLinkedQueue<ByteBuffer> _outbox;
+ private final EventListenerList _eventListeners;
+ private final int _deliveryTimeoutSec;
+ private final MessageFactory _msgFactory;
+
+ public ProtonClient(ProtonReactor reactor, Session session) {
+ _ssn = session;
+ _sender = null;
+ _receiver = null;
+ _outbox = new ConcurrentLinkedQueue<>();
+ _eventListeners = new EventListenerList();
+ _deliveryTimeoutSec = 60 * 3;
+ _reactor = reactor;
+ _msgFactory = new MessageFactoryImpl();
+ }
+
+ @Override
+ public void addEventListener(EventListener el) {
+ synchronized (_eventListeners) {
+ _eventListeners.add(EventListener.class, el);
+ }
+ }
+
+ @Override
+ public void removeEventListener(EventListener el) {
+ synchronized (_eventListeners) {
+ _eventListeners.remove(EventListener.class, el);
+ }
+ }
+
+ private void emitOnMessageReceived(ByteBuffer message) {
+ synchronized (_eventListeners) {
+ final Class<EventListener> cls = EventListener.class;
+ final EventListener[] els = _eventListeners.getListeners(cls);
+ for (EventListener el : els) {
+ el.onMessageReceived(this, message);
+ }
+ }
+ }
+
+ @Override
+ public void sendMessage(ByteBuffer message) {
+ _outbox.add(message);
+ _reactor.wakeup();
+ }
+
+ public void addLink(Link link) {
+ assert (link.getSession().equals(_ssn));
+
+ if (link instanceof Sender) {
+ if (_sender != null) {
+ // already have a sender
+ link.close();
+ return;
+ }
+
+ _sender = (Sender) link;
+ } else {
+ assert (link instanceof Receiver);
+ if (_receiver != null) {
+ // already have a receiver
+ link.close();
+ return;
+ }
+
+ _receiver = (Receiver) link;
+ _receiver.flow(_CREDIT);
+ }
+ link.open();
+ }
+
+ private Message _popOutgoingMessage() {
+ final ByteBuffer data = _outbox.poll();
+ if (data == null) {
+ return null;
+ }
+
+ final Section body = new Data(Binary.create(data));
+ final Message msg = _msgFactory.createMessage();
+ msg.setBody(body);
+ msg.setAddress(_sender.getTarget().toString());
+ return msg;
+ }
+
+ public void queueDeliveries() {
+ if (_sender == null) {
+ final String uuid = UUID.randomUUID().toString();
+ _sender = _ssn.sender("Sender-" + uuid);
+ }
+
+ while (_sender.getCredit() > 0) {
+ final Message m = _popOutgoingMessage();
+ if (m == null) {
+ return;
+ }
+
+ final String uuid = UUID.randomUUID().toString();
+ final Delivery d = _sender
+ .delivery(("outgoing-" + uuid).getBytes());
+ d.setContext(m);
+ }
+ }
+
+ public void processDelivery(Delivery delivery) {
+ assert (_ssn.equals(delivery.getLink().getSession()));
+
+ if (delivery.isReadable()) {
+ _processIncomingDelivery(delivery);
+ } else {
+ assert (delivery.isWritable());
+ _processOutgoingDelivery(delivery);
+ }
+ }
+
+ private void _processOutgoingDelivery(Delivery delivery) {
+ final Sender link = (Sender) delivery.getLink();
+ assert (link.equals(_sender));
+
+ final Message msg = (Message) delivery.getContext();
+ // TBD: Buffer can be reused forever. Change in case of
+ // performance issues.
+ ByteBuffer buff;
+ int i = 1;
+ int written = 0;
+ do {
+ buff = ByteBuffer.allocate(i * 4096);
+ written = msg.encode(buff.array(), 0, buff.capacity());
+ i++;
+ } while (written == buff.capacity());
+
+ link.send(buff.array(), 0, written);
+ if (link.advance()) {
+ // Attach timeout to the delivery
+ final Calendar calendar = Calendar.getInstance();
+ calendar.add(Calendar.SECOND, _deliveryTimeoutSec);
+ delivery.setContext(calendar);
+ }
+ }
+
+ private void _processIncomingDelivery(Delivery delivery) {
+ int total = 0;
+ int read = 0;
+ ByteBuffer buff = ByteBuffer.allocate(4096);
+
+ while (read >= 0) {
+ total += read;
+ if (total >= buff.capacity()) {
+ final ByteBuffer buff2 = ByteBuffer
+ .allocate(buff.capacity() * 2);
+ buff2.put(buff);
+ buff = buff2;
+ }
+ read = _receiver.recv(buff.array(), total, buff.capacity() - total);
+ }
+
+ final Message msg = _msgFactory.createMessage();
+ msg.decode(buff.array(), 0, total);
+
+ assert (msg.getBody() instanceof Data);
+ final Data body = (Data) msg.getBody();
+ final ByteBuffer bb = body.getValue().asByteBuffer();
+ delivery.settle();
+ emitOnMessageReceived(bb);
+ }
+
+ @Override
+ public Future<Void> close() {
+ final Session ssn = _ssn;
+ return _reactor.queueOperation(new Callable<Void>() {
+ @Override
+ public Void call() {
+ ssn.close();
+ return null;
+ }
+ });
+ }
+
+ @Override
+ public boolean closed() {
+ return _ssn.getLocalState().equals(EndpointState.CLOSED);
+ }
+
+ public void removeLink(Link link) {
+ if (link.equals(_sender)) {
+ _sender = null;
+ } else {
+ assert (link.equals(_receiver));
+ _receiver = null;
+ }
+ link.close();
+ }
+}
diff --git
a/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonListener.java
b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonListener.java
new file mode 100644
index 0000000..35896f4
--- /dev/null
+++ b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonListener.java
@@ -0,0 +1,42 @@
+package org.ovirt.vdsm.reactors;
+
+import java.io.IOException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+
+import org.apache.qpid.proton.driver.Listener;
+
+public final class ProtonListener implements ReactorListener {
+ private final EventListener _eventListener;
+ private Listener<ProtonListener> _listener;
+ private final ProtonReactor _reactor;
+
+ public ProtonListener(ProtonReactor reactor, EventListener eventListener) {
+ _eventListener = eventListener;
+ _reactor = reactor;
+ }
+
+ public void setListener(Listener<ProtonListener> l) {
+ _listener = l;
+ }
+
+ public void accept(ReactorClient client) {
+ _eventListener.onAcccept(this, client);
+ }
+
+ @Override
+ public Future<Void> close() {
+ final Listener<ProtonListener> l = _listener;
+ return _reactor.queueOperation(new Callable<Void>() {
+ @Override
+ public Void call() {
+ try {
+ l.close();
+ } catch (IOException e) {
+ // already closed
+ }
+ return null;
+ }
+ });
+ }
+}
diff --git
a/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonReactor.java
b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonReactor.java
new file mode 100644
index 0000000..b5a38b4
--- /dev/null
+++ b/client/java/vdsm-json-rpc/src/main/java/org/ovirt/vdsm/reactors/ProtonReactor.java
@@ -0,0 +1,452 @@
+package org.ovirt.vdsm.reactors;
+
+import java.io.IOException;
+import java.util.Calendar;
+import java.util.EnumSet;
+import java.util.Iterator;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+
+import org.apache.qpid.proton.driver.Connector;
+import org.apache.qpid.proton.driver.Driver;
+import org.apache.qpid.proton.driver.Listener;
+import org.apache.qpid.proton.driver.impl.DriverFactoryImpl;
+import org.apache.qpid.proton.engine.Connection;
+import org.apache.qpid.proton.engine.Delivery;
+import org.apache.qpid.proton.engine.EndpointState;
+import org.apache.qpid.proton.engine.EngineFactory;
+import org.apache.qpid.proton.engine.Link;
+import org.apache.qpid.proton.engine.Receiver;
+import org.apache.qpid.proton.engine.Session;
+import org.apache.qpid.proton.engine.impl.EngineFactoryImpl;
+import org.ovirt.vdsm.reactors.ProtonAuthenticator.AuthenticatorType;
+import org.ovirt.vdsm.reactors.ProtonAuthenticator.ConnectionState;
+import org.ovirt.vdsm.util.ChainedOperation;
+import org.ovirt.vdsm.util.ReactorScheduler;
+
+public final class ProtonReactor implements Reactor {
+
+ private final Driver _driver;
+ private final ReactorScheduler _scheduler;
+ private boolean _isRunning;
+ final Object _syncRoot = new Object();
+ final ProtonReactor reactor = this;
+ private EngineFactory _engineFactory;
+
+ public boolean isRunning() {
+ return _isRunning;
+ }
+
+ public ProtonReactor() throws IOException {
+ _engineFactory = new EngineFactoryImpl();
+ _driver = new DriverFactoryImpl().createDriver();
+ _isRunning = false;
+ _scheduler = new ReactorScheduler();
+ }
+
+ @Override
+ public void finalize() throws Throwable {
+ try {
+ _driver.destroy();
+ } finally {
+ super.finalize();
+ }
+ }
+
+ // Creates a listener, returns null if failed to bind or reactor is not
+ // running;
+ @Override
+ public Future<ReactorListener> createListener(final String host,
+ final int port,
+ final ReactorListener.EventListener eventListener) {
+
+ return queueOperation(new Callable<ReactorListener>() {
+ @Override
+ public ProtonListener call() {
+
+ final ProtonListener listener = new ProtonListener(reactor, eventListener);
+ final Listener<ProtonListener> l = _driver.createListener(host,
+ port, listener);
+
+ listener.setListener(l);
+
+ if (l == null) {
+ return null;
+ }
+
+ return listener;
+ }
+ });
+ }
+
+ @Override
+ public Future<ReactorClient> createClient(final String host, final int port) {
+ final Driver driver = _driver;
+ final EngineFactory engineFactory = _engineFactory;
+
+ return queueOperation(new ChainedOperation.Operation<ReactorClient>() {
+ final private int _INIT = 1;
+ final private int _AUTHENTICATE = 2;
+ final private int _DONE = 3;
+ private int _state;
+ final private Driver _driver;
+ final private ProtonReactor _reactor;
+ private Connector<ProtonAuthenticator> _connector;
+ private ProtonAuthenticator _auth;
+ private boolean _done;
+ private boolean _cancelled;
+ private ReactorClient _result;
+ private EngineFactory _engineFactory;
+
+ {
+ _driver = driver;
+ _reactor = reactor;
+ _state = _INIT;
+ _done = false;
+ _cancelled = false;
+ _engineFactory = engineFactory;
+ }
+
+ @Override
+ public void call(final boolean cancelled) {
+ switch (_state) {
+ case _INIT:
+ if (cancelled) {
+ _cancelled = true;
+ _done = true;
+ return;
+ }
+
+ _connector = this._driver.createConnector(host, port, null);
+
+ final Connection connection = engineFactory.createConnection();
+ _connector.setConnection(connection);
+ _auth = new ProtonAuthenticator(_connector,
+ AuthenticatorType.CLIENT);
+ _connector.setContext(_auth);
+ connection.open();
+ _state = _AUTHENTICATE;
+ case _AUTHENTICATE:
+ if (cancelled) {
+ _cancelled = true;
+ _close();
+ return;
+ }
+
+ switch (_auth.getState()) {
+ case AUTHENTICATING:
+ _auth.authenticate();
+ try {
+ _connector.process();
+ } catch (IOException e) {
+ // ignore
+ }
+ return;
+ case FAILED:
+ _close();
+ return;
+ case CONNECTED:
+ // Success !
+ break;
+ }
+
+ Session ssn = _connector.getConnection().session();
+ ssn.open();
+ _result = new ProtonClient(_reactor, ssn);
+ ssn.setContext(_result);
+ _done = true;
+ _state = _DONE;
+ }
+ }
+
+ private void _close() {
+ _connector.getConnection().close();
+ _connector.close();
+ _done = true;
+ _result = null;
+ }
+
+ @Override
+ public boolean isDone() {
+ return _done;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return _cancelled;
+ }
+
+ @Override
+ public ReactorClient getResult() {
+ return _result;
+ }
+ });
+ }
+
+ // Queues operation to be run in the serving loop.
+ public <T> Future<T> queueOperation(Callable<T> cb) {
+ final FutureTask<T> task = new FutureTask<>(cb);
+ _queueFuture(task);
+ return task;
+ }
+
+ public <T> Future<T> queueOperation(ChainedOperation.Operation<T> op)
{
+ final ChainedOperation<T> task = new ChainedOperation<>(op);
+ _queueFuture(task);
+ return task;
+ }
+
+ private void _queueFuture(Future<?> op) {
+ synchronized (_scheduler) {
+ _scheduler.queueFuture(op);
+ wakeup();
+ }
+ }
+
+ private void _waitEvents() {
+ _driver.doWait(0);
+ }
+
+ public void wakeup() {
+ _driver.wakeup();
+ }
+
+ @Override
+ public void serve() {
+ synchronized (_syncRoot) {
+ _isRunning = true;
+ }
+
+ while (_isRunning) {
+ //_waitEvents();
+ synchronized (_scheduler) {
+ _scheduler.performPendingOperations();
+ }
+ _acceptConnectionRequests();
+ _processConnectors();
+ }
+ }
+
+ private void _processConnectors() {
+ for (Connector<?> connector = _driver.connector(); connector != null; connector =
_driver
+ .connector()) {
+ if (connector.isClosed()) {
+ connector.destroy();
+ continue;
+ }
+
+ try {
+ connector.process();
+ } catch (IOException e) {
+ continue;
+ }
+
+ final Object ctx = connector.getContext();
+ assert (ctx instanceof ProtonAuthenticator);
+
+ if (ctx instanceof ProtonAuthenticator) {
+ final ProtonAuthenticator auth = (ProtonAuthenticator) ctx;
+ ConnectionState cs = auth.getState();
+ if (cs.equals(ConnectionState.AUTHENTICATING)) {
+ auth.authenticate();
+ cs = auth.getState();
+ }
+
+ if (cs.equals(ConnectionState.CONNECTED)) {
+ if (connector.getConnection() == null) {
+ connector.setConnection(_engineFactory.createConnection());
+ }
+ _processConnector(connector);
+ }
+ }
+
+ try {
+ connector.process();
+ } catch (IOException e) {
+ continue;
+ }
+ }
+ }
+
+ private void _processConnector(Connector<?> connector) {
+ _initConnection(connector);
+ _openPendingSessions(connector);
+ _openLinks(connector);
+ _queueOutgoingDeliveries(connector);
+ _processDeliveries(connector);
+ _cleanDeliveries(connector);
+ _cleanLinks(connector);
+ _cleanSessions(connector);
+ }
+
+ private void _cleanSessions(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+ final EnumSet<EndpointState> localState = EnumSet
+ .of(EndpointState.ACTIVE);
+ final EnumSet<EndpointState> remoteState = EnumSet
+ .of(EndpointState.CLOSED);
+
+ for (Session ssn = conn.sessionHead(localState, remoteState); ssn != null; ssn
+ .next(localState, remoteState)) {
+
+ ssn.close();
+ }
+ }
+
+ private void _cleanLinks(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+ final EnumSet<EndpointState> localState = EnumSet
+ .of(EndpointState.ACTIVE);
+ final EnumSet<EndpointState> remoteState = EnumSet
+ .of(EndpointState.CLOSED);
+
+ for (Link link = conn.linkHead(localState, remoteState); link != null; link = link
+ .next(localState, remoteState)) {
+
+ final ProtonClient ssn = _getClient(link.getSession());
+ ssn.removeLink(link);
+ }
+ }
+
+ private void _cleanDeliveries(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+ final EnumSet<EndpointState> localState = EnumSet
+ .of(EndpointState.ACTIVE);
+ final EnumSet<EndpointState> remoteState = EnumSet
+ .allOf(EndpointState.class);
+ for (Link link = conn.linkHead(localState, remoteState); link != null; link = link
+ .next(localState, remoteState)) {
+
+ if (link instanceof Receiver) {
+ // We settle all incoming deliveries upon receive
+ continue;
+ }
+
+ Delivery d;
+ final Calendar now = Calendar.getInstance();
+ for (Iterator<Delivery> iter = link.unsettled(); iter.hasNext();) {
+ d = iter.next();
+ Object ctx = d.getContext();
+ if (!(ctx instanceof Calendar)) {
+ // Has not been sent yet
+ continue;
+ }
+
+ final Calendar timeout = (Calendar) ctx;
+ boolean remoteClosed = link.getRemoteState().equals(
+ EndpointState.CLOSED);
+ boolean timedOut = now.after(timeout);
+ if (d.remotelySettled() || timedOut || remoteClosed) {
+ d.settle();
+ d.free();
+ }
+ }
+
+ }
+
+ }
+
+ private void _processDeliveries(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+ for (Delivery delivery = conn.getWorkHead(); delivery != null; delivery = delivery
+ .getWorkNext()) {
+
+ final ProtonClient client = _getClient(delivery.getLink()
+ .getSession());
+ client.processDelivery(delivery);
+ }
+ }
+
+ private void _queueOutgoingDeliveries(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+
+ final EnumSet<EndpointState> localState = EnumSet
+ .of(EndpointState.ACTIVE);
+ final EnumSet<EndpointState> remoteState = EnumSet
+ .allOf(EndpointState.class);
+
+ for (Session ssn = conn.sessionHead(localState, remoteState); ssn != null; ssn = ssn
+ .next(localState, remoteState)) {
+
+ final ProtonClient client = _getClient(ssn);
+ client.queueDeliveries();
+ }
+ }
+
+ private void _openLinks(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+ final EnumSet<EndpointState> localState = EnumSet
+ .of(EndpointState.UNINITIALIZED);
+ final EnumSet<EndpointState> remoteState = EnumSet
+ .allOf(EndpointState.class);
+ for (Link link = conn.linkHead(localState, remoteState); link != null; link = link
+ .next(localState, remoteState)) {
+
+ // configure the link
+ link.setSource(link.getRemoteSource());
+ link.setTarget(link.getRemoteTarget());
+
+ final ProtonClient client = _getClient(link.getSession());
+ client.addLink(link);
+ }
+ }
+
+ private ProtonClient _getClient(Session ssn) {
+ return (ProtonClient) ssn.getContext();
+ }
+
+ private void _openPendingSessions(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+ final EnumSet<EndpointState> localState = EnumSet
+ .of(EndpointState.UNINITIALIZED);
+ final EnumSet<EndpointState> remoteState = EnumSet
+ .allOf(EndpointState.class);
+
+ for (Session ssn = conn.sessionHead(localState, remoteState); ssn != null; ssn = ssn
+ .next(localState, remoteState)) {
+
+ final ProtonClient client = new ProtonClient(this, ssn);
+ ssn.setContext(client);
+ final Object ctx = connector.getContext();
+ assert (ctx instanceof ProtonAuthenticator);
+ ProtonAuthenticator auth = (ProtonAuthenticator) ctx;
+ if (auth.getAuthType() == AuthenticatorType.SERVER) {
+ ssn.open();
+ final ProtonListener l = (ProtonListener) ctx;
+ l.accept(client);
+ } else {
+ ssn.close();
+ }
+ }
+ }
+
+ private void _initConnection(Connector<?> connector) {
+ final Connection conn = connector.getConnection();
+ if (conn.getLocalState().equals(EndpointState.UNINITIALIZED)) {
+ conn.open();
+ }
+ }
+
+ private void _acceptConnectionRequests() {
+ for (final Listener<?> l : _driver.listeners()) {
+
+ @SuppressWarnings("unchecked")
+ final Connector<ProtonAuthenticator> connector =
(Connector<ProtonAuthenticator>) l
+ .accept();
+ if (connector == null) {
+ continue;
+ }
+ connector.setContext(new ProtonAuthenticator(connector,
+ AuthenticatorType.SERVER));
+ }
+ }
+
+ public void stop() {
+ synchronized (_syncRoot) {
+ _isRunning = false;
+ }
+
+ wakeup();
+ }
+}
\ No newline at end of file
diff --git
a/client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/AmqpReactorTestHelper.java
b/client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/AmqpReactorTestHelper.java
new file mode 100644
index 0000000..46d9cc3
--- /dev/null
+++
b/client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/AmqpReactorTestHelper.java
@@ -0,0 +1,18 @@
+package org.ovirt.vdsm.jsonrpc;
+
+import java.io.IOException;
+import org.ovirt.vdsm.reactors.ProtonReactor;
+import org.ovirt.vdsm.reactors.Reactor;
+
+public class AmqpReactorTestHelper implements ReactorTestHelper {
+ @Override
+ public Reactor createReactor() throws IOException {
+ return new ProtonReactor();
+ }
+
+ @Override
+ public String getUriScheme() {
+ return "amqp";
+ }
+
+}
diff --git
a/client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/TestJsonRpcClientAMQP.java
b/client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/TestJsonRpcClientAMQP.java
new file mode 100644
index 0000000..9e0c24c
--- /dev/null
+++
b/client/java/vdsm-json-rpc/src/test/java/org/ovirt/vdsm/jsonrpc/TestJsonRpcClientAMQP.java
@@ -0,0 +1,10 @@
+package org.ovirt.vdsm.jsonrpc;
+
+public class TestJsonRpcClientAMQP extends TestJsonRpcClient {
+
+ @Override
+ protected ReactorTestHelper getHelper() {
+ return new AmqpReactorTestHelper();
+ }
+
+}
--
To view, visit
http://gerrit.ovirt.org/15428
To unsubscribe, visit
http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I94c52e118cb63d7df84b89a9b93da7b9e477be91
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Saggi Mizrahi <smizrahi(a)redhat.com>