Index: trunk/src/main/java/omq/common/broker/Broker.java =================================================================== --- trunk/src/main/java/omq/common/broker/Broker.java (revision 73) +++ trunk/src/main/java/omq/common/broker/Broker.java (revision 74) @@ -162,15 +162,10 @@ public void bind(String reference, RemoteObject remote) throws RemoteException { - try { - remote.startRemoteObject(reference, this); - remoteObjs.put(reference, remote); - } catch (Exception e) { - throw new RemoteException(e); - } - } - - public void startTriggerEvent(String reference, RemoteObject remote) throws RemoteException { - try { - remote.startTriggerEvent(reference, this); + bind(reference, remote, environment); + } + + public void bind(String reference, RemoteObject remote, Properties env) throws RemoteException { + try { + remote.startRemoteObject(reference, this, env); remoteObjs.put(reference, remote); } catch (Exception e) { Index: trunk/src/main/java/omq/server/RemoteObject.java =================================================================== --- trunk/src/main/java/omq/server/RemoteObject.java (revision 73) +++ trunk/src/main/java/omq/server/RemoteObject.java (revision 74) @@ -58,11 +58,11 @@ } - public void startRemoteObject(String reference, Broker broker) throws Exception { + public void startRemoteObject(String reference, Broker broker, Properties env) throws Exception { this.broker = broker; - UID = reference; - multiQueue = UID + System.currentTimeMillis(); - env = broker.getEnvironment(); - - params = new HashMap>>(); + this.UID = reference; + this.multiQueue = UID + System.currentTimeMillis(); + this.env = env; + + this.params = new HashMap>>(); for (Method m : this.getClass().getMethods()) { List> list = new ArrayList>(); @@ -70,10 +70,10 @@ list.add(clazz); } - params.put(m.getName(), list); + this.params.put(m.getName(), list); } // Get num threads to use int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1")); - remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer()); + this.remoteWrapper = new RemoteWrapper(this, numThreads, broker.getSerializer()); startQueues(); @@ -227,16 +227,13 @@ // Declares and bindings - logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue); + logger.info("RemoteObject: " + UID + " declaring direct exchange: " + exchange + ", Queue: " + queue + ", Durable: " + durable); channel.exchangeDeclare(exchange, "direct"); channel.queueDeclare(queue, durable, false, false, null); channel.queueBind(queue, exchange, routingKey); + logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + multiExchange + ", Queue: " + multiQueue + ", Durable: " + durable); channel.exchangeDeclare(multiExchange, "fanout"); channel.queueDeclare(multiQueue, durable, false, false, null); channel.queueBind(multiQueue, multiExchange, ""); - - // Declare the event topic fanout - logger.info("RemoteObject: " + UID + " declaring fanout exchange: " + UID); - channel.exchangeDeclare(UID, "fanout"); // Declare a new consumer Index: trunk/src/test/java/omq/test/workspace/Info.java =================================================================== --- trunk/src/test/java/omq/test/workspace/Info.java (revision 74) +++ trunk/src/test/java/omq/test/workspace/Info.java (revision 74) @@ -0,0 +1,27 @@ +package omq.test.workspace; + +import java.io.Serializable; + +public class Info implements Serializable { + /** + * + */ + private static final long serialVersionUID = 1L; + private String id; + + public Info() { + } + + public Info(String id) { + this.id = id; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + +} Index: trunk/src/test/java/omq/test/workspace/RemoteWorkspace.java =================================================================== --- trunk/src/test/java/omq/test/workspace/RemoteWorkspace.java (revision 74) +++ trunk/src/test/java/omq/test/workspace/RemoteWorkspace.java (revision 74) @@ -0,0 +1,13 @@ +package omq.test.workspace; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.MultiMethod; +import omq.client.annotation.RemoteInterface; + +@RemoteInterface +public interface RemoteWorkspace extends Remote { + @MultiMethod + @AsyncMethod + public void update(Info info); +} Index: trunk/src/test/java/omq/test/workspace/RemoteWorkspaceImpl.java =================================================================== --- trunk/src/test/java/omq/test/workspace/RemoteWorkspaceImpl.java (revision 74) +++ trunk/src/test/java/omq/test/workspace/RemoteWorkspaceImpl.java (revision 74) @@ -0,0 +1,25 @@ +package omq.test.workspace; + +import omq.server.RemoteObject; + +public class RemoteWorkspaceImpl extends RemoteObject implements RemoteWorkspace { + + /** + * + */ + private static final long serialVersionUID = 1L; + private Info info; + + public void update(Info info) { + this.info = info; + } + + public Info getInfo() { + return info; + } + + public void setInfo(Info info) { + this.info = info; + } + +} Index: trunk/src/test/java/omq/test/workspace/WorkspaceTest.java =================================================================== --- trunk/src/test/java/omq/test/workspace/WorkspaceTest.java (revision 74) +++ trunk/src/test/java/omq/test/workspace/WorkspaceTest.java (revision 74) @@ -0,0 +1,127 @@ +package omq.test.workspace; + +import static org.junit.Assert.assertEquals; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; +import omq.common.util.Serializer; + +import org.junit.After; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; + +@RunWith(value = Parameterized.class) +public class WorkspaceTest { + private static Broker serverBroker; + private static Broker clientBroker1; + private static Broker clientBroker2; + private static String[] workspaces = { "w1", "w2", "w3" }; + private static RemoteWorkspace[] remoteWorks = new RemoteWorkspace[3]; + + // In this case the Constructor acts as a server + public WorkspaceTest(String type) throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Set host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.SERIALIZER_NAME, type); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + + serverBroker = new Broker(env); + int i = 0; + for (String w : workspaces) { + remoteWorks[i++] = serverBroker.lookupMulti(w, RemoteWorkspace.class); + } + + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @BeforeClass + public static void client() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.SERVER_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.SERVER_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.ENABLECOMPRESSION, "false"); + + // Set info about where the message will be sent + env.setProperty(ParameterQueue.RPC_EXCHANGE, "rpc_exchange"); + env.setProperty(ParameterQueue.RETRY_TIME_CONNECTION, "2000"); + + clientBroker1 = new Broker(env); + clientBroker2 = new Broker(env); + + System.out.println("Client 1 & client2 started"); + } + + @After + public void stop() throws Exception { + serverBroker.stopBroker(); + } + + @Test + public void test() throws Exception { + String expected = null; + String actual = null; + + // Client 1 will subscribe to changes in the workspaces w1 and w3 + RemoteWorkspaceImpl w1C1 = new RemoteWorkspaceImpl(); + RemoteWorkspaceImpl w3C1 = new RemoteWorkspaceImpl(); + + clientBroker1.bind(workspaces[0], w1C1); + clientBroker1.bind(workspaces[2], w3C1); + + // Client 2 will subscribe to changes in the workspaces w1 and w2 + RemoteWorkspaceImpl w1C2 = new RemoteWorkspaceImpl(); + RemoteWorkspaceImpl w2C2 = new RemoteWorkspaceImpl(); + + clientBroker2.bind(workspaces[0], w1C2); + clientBroker2.bind(workspaces[2], w2C2); + + // The server will notify a change in the w2 + expected = "w2 has changed"; + Info info = new Info(expected); + remoteWorks[1].update(info); + Thread.sleep(200); + + // If everything has worked, the client2 will see a change in w2 + actual = w2C2.getInfo().getId(); + assertEquals(expected, actual); + + // The server will notify a change in the w1 + expected = "w1 has changed"; + info.setId(expected); + remoteWorks[0].update(info); + Thread.sleep(200); + + // If everything has worked, both clients will see a change in w1 + actual = w1C1.getInfo().getId(); + assertEquals(expected, actual); + actual = w1C2.getInfo().getId(); + assertEquals(expected, actual); + } + +}