Index: trunk/src/test/java/omq/test/observer/ObserverTest.java =================================================================== --- trunk/src/test/java/omq/test/observer/ObserverTest.java (revision 71) +++ trunk/src/test/java/omq/test/observer/ObserverTest.java (revision 71) @@ -0,0 +1,102 @@ +package omq.test.observer; + +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 ObserverTest { + private static String SUBJECT = "subject"; + private static String OBSERVER = "observer"; + private static Broker broker; + private static RemoteSubject remoteSubject; + + public ObserverTest(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"); + + // Set info about the queue & the exchange where the ResponseListener + // will listen to. + env.setProperty(ParameterQueue.RPC_REPLY_QUEUE, "reply_queue"); + env.setProperty(ParameterQueue.EVENT_REPLY_QUEUE, "event_queue"); + + broker = new Broker(env); + remoteSubject = broker.lookup(SUBJECT, RemoteSubject.class); + } + + @Parameters + public static Collection data() { + Object[][] data = new Object[][] { { Serializer.JAVA }, { Serializer.GSON }, { Serializer.KRYO } }; + return Arrays.asList(data); + } + + @BeforeClass + public static void server() 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"); + + Broker broker = new Broker(env); + RemoteSubjectImpl subject = new RemoteSubjectImpl(); + broker.bind(SUBJECT, subject); + + System.out.println("Server started"); + } + + @After + public void stop() throws Exception { + broker.stopBroker(); + } + + @Test + public void test() throws Exception { + String expected = "I'm fine"; + String actual = null; + + RemoteObserverImpl observer = new RemoteObserverImpl(); + broker.bind(OBSERVER, observer); + + remoteSubject.addObserver(observer); + remoteSubject.setState(expected); + remoteSubject.notifyObservers(); + actual = observer.getObsState(); + + assertEquals(expected, actual); + remoteSubject.setState(null); + } + +} Index: trunk/src/test/java/omq/test/observer/RemoteObserver.java =================================================================== --- trunk/src/test/java/omq/test/observer/RemoteObserver.java (revision 71) +++ trunk/src/test/java/omq/test/observer/RemoteObserver.java (revision 71) @@ -0,0 +1,11 @@ +package omq.test.observer; + +import omq.Remote; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; + +@RemoteInterface +public interface RemoteObserver extends Remote { + @SyncMethod(timeout = 1000) + public void update(); +} Index: trunk/src/test/java/omq/test/observer/RemoteObserverImpl.java =================================================================== --- trunk/src/test/java/omq/test/observer/RemoteObserverImpl.java (revision 71) +++ trunk/src/test/java/omq/test/observer/RemoteObserverImpl.java (revision 71) @@ -0,0 +1,44 @@ +package omq.test.observer; + +import omq.server.RemoteObject; + +public class RemoteObserverImpl extends RemoteObject implements RemoteObserver { + + /** + * + */ + private static final long serialVersionUID = 1L; + private String obsState; + private RemoteSubject subject; + + @Override + public void update() { + obsState = subject.getStringState(); + } + + public String getObsState() { + return obsState; + } + + public void setObsState(String obsState) { + this.obsState = obsState; + } + + public RemoteSubject getSubject() { + return subject; + } + + public void setSubject(RemoteSubject subject) { + this.subject = subject; + } + + @Override + public boolean equals(Object obj) { + if (obj instanceof RemoteObserver) { + RemoteObserver ro = (RemoteObserver) obj; + this.getRef().endsWith(ro.getRef()); + } + return false; + } + +} Index: trunk/src/test/java/omq/test/observer/RemoteSubject.java =================================================================== --- trunk/src/test/java/omq/test/observer/RemoteSubject.java (revision 71) +++ trunk/src/test/java/omq/test/observer/RemoteSubject.java (revision 71) @@ -0,0 +1,24 @@ +package omq.test.observer; + +import omq.Remote; +import omq.client.annotation.RemoteInterface; +import omq.client.annotation.SyncMethod; + +@RemoteInterface +public interface RemoteSubject extends Remote { + + @SyncMethod(timeout = 1000) + public String getStringState(); + + @SyncMethod(timeout = 1000) + public void setState(String state); + + @SyncMethod(timeout = 1000) + public void addObserver(RemoteObserver o); + + @SyncMethod(timeout = 1000) + public void removeObserver(RemoteObserver o); + + @SyncMethod(timeout = 1000) + public void notifyObservers(); +} Index: trunk/src/test/java/omq/test/observer/RemoteSubjectImpl.java =================================================================== --- trunk/src/test/java/omq/test/observer/RemoteSubjectImpl.java (revision 71) +++ trunk/src/test/java/omq/test/observer/RemoteSubjectImpl.java (revision 71) @@ -0,0 +1,43 @@ +package omq.test.observer; + +import java.util.ArrayList; +import java.util.List; + +import omq.server.RemoteObject; + +public class RemoteSubjectImpl extends RemoteObject implements RemoteSubject { + + /** + * + */ + private static final long serialVersionUID = 1L; + private String state; + private List list = new ArrayList(); + + @Override + public void addObserver(RemoteObserver o) { + list.add(o); + } + + @Override + public void removeObserver(RemoteObserver o) { + list.remove(o); + } + + @Override + public void notifyObservers() { + for (RemoteObserver o : list) { + o.update(); + } + } + + @Override + public void setState(String state) { + this.state = state; + } + + @Override + public String getStringState() { + return state; + } +} Index: trunk/target/classes/log4j.xml =================================================================== --- trunk/target/classes/log4j.xml (revision 71) +++ trunk/target/classes/log4j.xml (revision 71) @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +