Change in vdsm[master]: [WIP]Java Bindings: Proton support in Java Bindings - vdsm-patches - Fedora mailing-lists

6 Jun 2013

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@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@redhat.com

    