Index: trunk/src/test/java/omq/test/persistence/PersistentTest.java =================================================================== --- trunk/src/test/java/omq/test/persistence/PersistentTest.java (revision 78) +++ trunk/src/test/java/omq/test/persistence/PersistentTest.java (revision 81) @@ -47,4 +47,5 @@ msgImplProps.setProperty(ParameterQueue.AUTO_DELETE_QUEUE, "false"); msgImplProps.setProperty(ParameterQueue.MULTI_QUEUE_NAME, "multiMessageQueue"); + msgImplProps.setProperty(ParameterQueue.DELIVERY_MODE, "1"); /* Index: trunk/src/test/java/omq/test/python/Address.java =================================================================== --- trunk/src/test/java/omq/test/python/Address.java (revision 81) +++ trunk/src/test/java/omq/test/python/Address.java (revision 81) @@ -0,0 +1,46 @@ +package omq.test.python; + +import java.io.Serializable; + +public class Address implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private String street; + private int number; + private String town; + + public Address(String street, int number, String town) { + this.street = street; + this.number = number; + this.town = town; + } + + public String getStreet() { + return street; + } + + public void setStreet(String street) { + this.street = street; + } + + public int getNumber() { + return number; + } + + public void setNumber(int number) { + this.number = number; + } + + public String getTown() { + return town; + } + + public void setTown(String town) { + this.town = town; + } + +} Index: trunk/src/test/java/omq/test/python/Calculator.java =================================================================== --- trunk/src/test/java/omq/test/python/Calculator.java (revision 81) +++ trunk/src/test/java/omq/test/python/Calculator.java (revision 81) @@ -0,0 +1,12 @@ +package omq.test.python; + +import omq.Remote; + +public interface Calculator extends Remote { + public void fibonacci(int x); + + public int add(int x, int y); + + public double getPi(); + +} Index: trunk/src/test/java/omq/test/python/CalculatorImpl.java =================================================================== --- trunk/src/test/java/omq/test/python/CalculatorImpl.java (revision 81) +++ trunk/src/test/java/omq/test/python/CalculatorImpl.java (revision 81) @@ -0,0 +1,39 @@ +package omq.test.python; + +import omq.server.RemoteObject; + +public class CalculatorImpl extends RemoteObject implements Calculator { + + /** + * + */ + private static final long serialVersionUID = 1L; + + @Override + public void fibonacci(int x) { + System.out.println(fib(x)); + } + + public int fib(int x) { + if (x == 0) { + return 0; + } + if (x == 1) { + return 1; + } + return fib(x - 1) + fib(x - 2); + } + + public int add(int x, int y) { + System.out.println("add: " + x + "+" + y + "=" + (x + y)); + return x + y; + } + + @Override + public double getPi() { + return 3.1415; + } + + + +} Index: trunk/src/test/java/omq/test/python/Client.py =================================================================== --- trunk/src/test/java/omq/test/python/Client.py (revision 81) +++ trunk/src/test/java/omq/test/python/Client.py (revision 81) @@ -0,0 +1,101 @@ +''' +Created on 03/07/2013 + +@author: sergi +''' +import pika +import uuid +import json + +class Client(object): + ''' + classdocs + ''' + + + def __init__(self, env): + ''' + Constructor + ''' + self.rpc_exchange = env.get('exchange', 'rpc_exchange') + self.reply_queue = env.get('reply_queue', 'reply_queue') + + self.host = env.get('host', 'localhost') + self.port = env.get('port', 5672) + self.ssl = env.get('ssl', False) + + self.credentials = pika.PlainCredentials(env.get('user', 'guest'), env.get('pass', 'guest')) + + self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, port=self.port, credentials=self.credentials, ssl=self.ssl)) + + self.channel = self.connection.channel() + + self.callback_queue = self.channel.queue_declare(queue=self.reply_queue) + + self.channel.basic_consume(self.on_response, no_ack=True, queue=self.reply_queue) + + def async_call(self, uid, method, params): + """ Async call this function will invoke the method 'method' with the params 'params' in the object binded with 'uid'""" + self.__async(uid, method, params, False) + + def multi_async_call(self, uid, method, params): + self.__async(uid, method, params, True) + + def __async(self, uid, method, params, multi): + if multi: + exch = "multi#" + uid + else: + exch = self.rpc_exchange + + corr_id = str(uuid.uuid4()) + request = json.dumps({"method": method, "params": params, "id": corr_id, "async":"true"}) + + props = pika.BasicProperties(app_id=uid, correlation_id=corr_id, reply_to="", type='gson') + + self.channel.basic_publish(exchange=exch, routing_key=uid, properties=props, body=request) + + print "UID: " + uid + ", exchange: " + exch + " ,corrId = " + corr_id + " , json: " + request + + def sync_call(self, uid, method, params): + self.response = None + self.corr_id = str(uuid.uuid4()) + request = json.dumps({"method": method, "params": params, "id": self.corr_id, "async":"false"}) + + props = pika.BasicProperties(app_id=uid, correlation_id=self.corr_id, reply_to="reply_queue", type='gson') + + self.channel.basic_publish(exchange=self.rpc_exchange, routing_key=uid, properties=props, body=request) + + print "UID: " + uid + ", exchange: " + self.rpc_exchange + " ,corrId = " + self.corr_id + " , json: " + request + + return self.__get_response() + + def multi_sync_call(self, uid, method, params, wait): + responses = [] + self.response = None + self.corr_id = str(uuid.uuid4()) + rpc_exchange = "multi#" + uid + request = json.dumps({"method": method, "params": params, "id": self.corr_id, "async":"false"}) + + props = pika.BasicProperties(app_id=uid, correlation_id=self.corr_id, reply_to="reply_queue", type='gson') + + self.channel.basic_publish(exchange=rpc_exchange, routing_key=uid, properties=props, body=request) + + print "UID: " + uid + ", exchange: " + rpc_exchange + " ,corrId = " + self.corr_id + " , json: " + request + i = 0 + while i < wait: + responses.append(self.__get_response()) + self.response = None + i = i + 1 + return responses + + def __get_response(self): + while self.response is None: + self.connection.process_data_events() + return self.response + + def on_response(self, ch, method, props, body): + if self.corr_id == props.correlation_id: + result = json.loads(body) + self.response = result["result"] + + Index: trunk/src/test/java/omq/test/python/ClientTest.py =================================================================== --- trunk/src/test/java/omq/test/python/ClientTest.py (revision 81) +++ trunk/src/test/java/omq/test/python/ClientTest.py (revision 81) @@ -0,0 +1,61 @@ +''' +Created on 04/07/2013 + +@author: sergi +''' +import unittest +from Client import Client +import time + +class Test(unittest.TestCase): + env = {'user': 'guest', 'pass':'guest', 'host' : 'localhost', 'port': 5672, 'exchange': 'rpc_exchange'} + + client = Client(env) + + def testAsync(self): + self.client.async_call("calculator", "fibonacci", [10]) + + def testSyncAdd(self): + x = 10 + y = 5 + expected = x + y + actual = self.client.sync_call("calculator", "add", [x, y]) + self.assertEqual(expected, actual) + + def testMultiSyncAdd(self): + x = 20 + y = 10 + expected = x + y + actual = self.client.multi_sync_call("calculator", "add", [x, y], 2) + self.assertEqual(expected, actual[0]) + self.assertEqual(expected, actual[1]) + + def testPi(self): + expected = 3.1415 + actual = self.client.sync_call("calculator", "getPi", []) + self.assertEqual(expected, actual) + + def testContact(self): + address = {"street":"calle falsa", "number":2, "town":"Tgn"} + emails = ["asdf@gmail.com", "asdf@hotmail.com"] + + name1 = "Superman" + contact = {"name":name1, "surname":"T", "phone":"1234", "age":21, "address": address, "emails":emails} + self.client.async_call("list", "setContact", [contact]) + time.sleep(1) + + name2 = "Batman" + contact = {"name":name2, "surname":"S", "phone":"1234", "age":22, "address": address, "emails":emails} + self.client.async_call("list", "setContact", [contact]) + time.sleep(1) + + clist = self.client.sync_call("list", "getContacts", []) + c1 = clist[0] + c2 = clist[1] + + self.assertEqual(name1, c1['name']) + self.assertEqual(name2, c2['name']) + +if __name__ == "__main__": + # import sys;sys.argv = ['', 'Test.testName'] + unittest.main() Index: trunk/src/test/java/omq/test/python/Contact.java =================================================================== --- trunk/src/test/java/omq/test/python/Contact.java (revision 81) +++ trunk/src/test/java/omq/test/python/Contact.java (revision 81) @@ -0,0 +1,81 @@ +package omq.test.python; + +import java.io.Serializable; + +public class Contact implements Serializable { + + /** + * + */ + private static final long serialVersionUID = 1L; + + private String name; + private String surname; + private String phone; + private int age; + private Address address; + + private String[] emails; + + public Contact(String name, String surname, String phone, int age, Address address, String[] emails) { + this.name = name; + this.surname = surname; + this.phone = phone; + this.age = age; + this.address = address; + this.emails = emails; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getSurname() { + return surname; + } + + public void setSurname(String surname) { + this.surname = surname; + } + + public String getPhone() { + return phone; + } + + public void setPhone(String phone) { + this.phone = phone; + } + + public int getAge() { + return age; + } + + public void setAge(int age) { + this.age = age; + } + + public Address getAddress() { + return address; + } + + public void setAddress(Address address) { + this.address = address; + } + + public String[] getEmails() { + return emails; + } + + public void setEmails(String[] emails) { + this.emails = emails; + } + + @Override + public String toString() { + return "Name: " + this.name + ", phone: " + this.phone; + } +} Index: trunk/src/test/java/omq/test/python/ContactList.java =================================================================== --- trunk/src/test/java/omq/test/python/ContactList.java (revision 81) +++ trunk/src/test/java/omq/test/python/ContactList.java (revision 81) @@ -0,0 +1,11 @@ +package omq.test.python; + +import java.util.List; + +import omq.Remote; + +public interface ContactList extends Remote { + public void setContact(Contact c); + + public List getContacts(); +} Index: trunk/src/test/java/omq/test/python/ContactListImpl.java =================================================================== --- trunk/src/test/java/omq/test/python/ContactListImpl.java (revision 81) +++ trunk/src/test/java/omq/test/python/ContactListImpl.java (revision 81) @@ -0,0 +1,25 @@ +package omq.test.python; + +import java.util.ArrayList; +import java.util.List; + +import omq.server.RemoteObject; + +public class ContactListImpl extends RemoteObject implements ContactList { + /** + * + */ + private static final long serialVersionUID = 1L; + private List list = new ArrayList(); + + @Override + public void setContact(Contact c) { + list.add(c); + System.out.println("New contact added: " + c); + } + + @Override + public List getContacts() { + return list; + } +} Index: trunk/src/test/java/omq/test/python/Server.java =================================================================== --- trunk/src/test/java/omq/test/python/Server.java (revision 81) +++ trunk/src/test/java/omq/test/python/Server.java (revision 81) @@ -0,0 +1,41 @@ +package omq.test.python; + +import java.util.Properties; + +import org.junit.Test; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +public class Server { + + @Test + public void test() 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.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + env.setProperty(ParameterQueue.DURABLE_QUEUES, "false"); + env.setProperty(ParameterQueue.ENABLE_COMPRESSION, "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); + + broker.bind("calculator", new CalculatorImpl()); + + Broker broker2 = new Broker(env); + + broker2.bind("calculator", new CalculatorImpl()); + + broker.bind("list", new ContactListImpl()); + + Thread.sleep(60 * 1000); + } + +}