raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

added optional routingkey and queuename to parameters

Browse source
This commit is contained in:
momania 2010-08-09 10:28:19 +02:00
parent 6e6f1f32d2
commit 32ece975c5
1 changed files with 14 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

25
akka-amqp/src/main/scala/se/scalablesolutions/akka/amqp/rpc/RPC.scala
View file

@ -61,7 +61,9 @@ object RPC {
private val ARRAY_OF_BYTE_ARRAY = Array[Class[_]](classOf[Array[Byte]])
def startProtobufServer[I <: Message, O <: Message](
connection: ActorRef, serviceName: String, requestHandler: I => O)(implicit manifest: Manifest[I]) = {
connection: ActorRef, exchange: String, requestHandler: I => O,
routingKey: Option[String] = None,
queueName: Option[String] = None)(implicit manifest: Manifest[I]) = {
val serializer = new RpcServerSerializer[I, O](
new FromBinary[I] {
@ -72,16 +74,18 @@ object RPC {
def toBinary(t: O) = t.toByteArray
})
val exchangeParameters = new ExchangeParameters(serviceName, ExchangeType.Topic)
val routingKey = "%s.request".format(serviceName)
val queueName = "%s.in".format(routingKey)
val exchangeParameters = new ExchangeParameters(exchange, ExchangeType.Topic)
val rKey = routingKey.getOrElse("%s.request".format(exchange))
val qName = queueName.getOrElse("%s.in".format(rKey))
newRpcServer[I, O](connection, exchangeParameters, routingKey, serializer, requestHandler,
queueName = Some(queueName))
newRpcServer[I, O](connection, exchangeParameters, rKey, serializer, requestHandler,
queueName = Some(qName))
}
def startProtobufClient[O <: Message, I <: Message](
connection: ActorRef, serviceName: String)(implicit manifest: Manifest[I]): RpcClient[O, I] = {
connection: ActorRef, exchange: String,
routingKey: Option[String] = None)(implicit manifest: Manifest[I]): RpcClient[O, I] = {
val serializer = new RpcClientSerializer[O, I](
new ToBinary[O] {
@ -92,11 +96,10 @@ object RPC {
}
})
val exchangeParameters = new ExchangeParameters(serviceName, ExchangeType.Topic)
val routingKey = "%s.request".format(serviceName)
val queueName = "%s.in".format(routingKey)
val exchangeParameters = new ExchangeParameters(exchange, ExchangeType.Topic)
val rKey = routingKey.getOrElse("%s.request".format(exchange))
val client = newRpcClient[O, I](connection, exchangeParameters, routingKey, serializer)
val client = newRpcClient[O, I](connection, exchangeParameters, rKey, serializer)
new RpcClient[O, I](client)
}