finagle-protobuf

finagle-protobuf/.gitignore

@@ -0,0 +1,3 @@
+.classpath
+.project
+.settings

finagle-protobuf/pom.xml

@@ -0,0 +1,38 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<groupId>com.twitter</groupId>
+	<artifactId>finagle-protobuf</artifactId>
+	<packaging>jar</packaging>
+	<version>6.5.3-TENDRIL</version>
+	<distributionManagement>
+		<repository>
+			<id>artifactory-release</id>
+			<name>Artifactory Releases</name>
+			<url>http://build:8081/nexus/content/repositories/platform-services</url>
+		</repository>
+		<snapshotRepository>
+			<id>artifactory-snapshot</id>
+			<name>Artifactory Snapshots</name>
+			<url>http://build:8081/nexus/content/repositories/platform-services-snapshot</url>
+		</snapshotRepository>
+	</distributionManagement>
+	<scm>
+       <connection>scm:git:https://fisheye.tendrilinc.com/git/finagle.git</connection>
+       <developerConnection>scm:git:https://fisheye.tendrilinc.com/git/finagle.git</developerConnection>
+    </scm>
+    <build>
+        <directory>target/scala-2.9.2</directory>
+    </build>   
+	<dependencies>
+		<dependency>
+			<groupId>com.twitter</groupId>
+			<artifactId>finagle-core</artifactId>
+			<version>3.0.0</version>
+		</dependency>
+		<dependency>
+			<groupId>com.google.protobuf</groupId>
+			<artifactId>protobuf-java</artifactId>
+			<version>2.4.1</version>
+		</dependency>
+	</dependencies>
+</project>

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/ExceptionResponseHandler.scala

@@ -0,0 +1,12 @@
+package com.twitter.finagle.protobuf.rpc
+
+/**
+ * Inspects a response message and if it contains an exception code it throws the correspondent exception. See also ServiceExceptionHandler.
+ */
+trait ExceptionResponseHandler[T] {
+
+  def canHandle(message: T): Boolean
+
+  def handle(message: T): RuntimeException
+
+}

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/RpcControllerWithOnFailureCallback.scala

@@ -1,5 +1,10 @@
 package com.twitter.finagle.protobuf.rpc
 
+import com.google.protobuf.RpcCallback
+import com.google.protobuf.RpcController
+import com.twitter.finagle.TimeoutException
+import com.twitter.finagle.ChannelClosedException
+
 class RpcControllerWithOnFailureCallback extends RpcController {
 
   private var cancelRequested = false

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/RpcFactory.scala

@@ -1,15 +1,22 @@
 package com.twitter.finagle.protobuf.rpc
 
+import com.google.protobuf.Message
+import com.google.protobuf.RpcChannel
+import com.google.protobuf.Service
+import com.twitter.finagle.builder.ClientBuilder
+import com.twitter.finagle.builder.ServerBuilder
+import com.google.protobuf.RpcController
+import com.twitter.util.Duration
 import java.util.concurrent.ExecutorService
 
 trait RpcFactory {
 
-  def createServer(sb: ServerBuilder[(String, Message), (String, Message), Any, Any, Any], port: Int, service: Service, executorService: ExecutorService): RpcServer
+  def createServer(sb: ServerBuilder[(String, Message), (String, Message), Any, Any, Any], port: Int, service: Service, handler: ServiceExceptionHandler[Message], executorService: ExecutorService): RpcServer
 
-  def createStub[T <: Service](cb: ClientBuilder[(String, Message), (String, Message), Any, Any, Any], service: { def newStub(c: RpcChannel): T }, executorService: ExecutorService): T
+  def createStub[T <: Service](cb: ClientBuilder[(String, Message), (String, Message), Any, Any, Any], service: {def newStub(c: RpcChannel): T}, handler: ExceptionResponseHandler[Message], executorService: ExecutorService): T
 
   def createController(): RpcController
 
-  def release(stub: { def getChannel(): RpcChannel })
+  def release(stub: {def getChannel(): RpcChannel})
 
 }

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/RpcServer.scala

@@ -1,5 +1,7 @@
 package com.twitter.finagle.protobuf.rpc
 
+import com.twitter.util.Duration
+
 trait RpcServer {
 
   def close(d: Duration): Unit;

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/ServiceExceptionHandler.scala

@@ -0,0 +1,15 @@
+package com.twitter.finagle.protobuf.rpc
+
+import com.google.protobuf.Message
+
+/**
+ * If the invocation of a service causes a RuntimeException this handler serializes it into a response message
+ * that can be deserialized by an ExceptioResponseHandler.
+ */
+trait ServiceExceptionHandler[T] {
+
+  def canHandle(e: RuntimeException): Boolean
+
+  def handle(e: RuntimeException, m: Message): T
+
+}

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/Util.scala

@@ -2,18 +2,35 @@ package com.twitter.finagle.protobuf.rpc
 
 import java.util.List
 
+import com.google.common.base.Function
+import com.google.common.collect.Lists
+import com.google.protobuf.Descriptors.MethodDescriptor
+import com.google.protobuf.Service
+import com.google.protobuf.Message
+
+import scala.collection.JavaConversions._
+import scala.collection.mutable._
+
+import org.slf4j.LoggerFactory
 
 object Util {
 
-	def extractMethodNames(s: Service) : List[String] = {
-		return Lists.transform(s.getDescriptorForType().getMethods(),
-				new Function[MethodDescriptor, String]() {
+  private val log = LoggerFactory.getLogger(getClass)
+
+  def extractMethodNames(s: Service): List[String] = {
+    return Lists.transform(s.getDescriptorForType().getMethods(),
+      new Function[MethodDescriptor, String]() {
 
-					@Override
-					def apply(d: MethodDescriptor) : String = {
-						return d.getName()
-					}
-				})
-	}
+        @Override
+        def apply(d: MethodDescriptor): String = {
+          return d.getName()
+        }
+      })
+  }
 
+  def log(reqOrResp: String, method: String, m: Message) {
+    if (log.isDebugEnabled()) {
+      log.debug("#log# {} {}: {}", Array[AnyRef](reqOrResp, method, m.toString()))
+    }
+  }
 }

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/channel/ClientSideDecoder.scala

@@ -1,5 +1,14 @@
 package com.twitter.finagle.protobuf.rpc.channel
 
+import org.jboss.netty.buffer.ChannelBuffer
+import org.jboss.netty.buffer.ChannelBufferInputStream
+import org.jboss.netty.channel.Channel
+import org.jboss.netty.channel.ChannelHandlerContext
+import org.jboss.netty.handler.codec.frame.FrameDecoder
+
+import com.google.protobuf.Message
+import com.google.protobuf.Service
+
 
 class ClientSideDecoder(val repo: MethodLookup, val service: Service) extends FrameDecoder with ProtobufDecoder {
 

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/channel/MethodLookup.scala

@@ -2,12 +2,12 @@ package com.twitter.finagle.protobuf.rpc.channel
 
 /**
  * Lookup repository that maps method names to 4 bytes.
- * 
- * */
+ *
+ **/
 trait MethodLookup {
 
-  	def encode(methodName: String ) : Int
+  def encode(methodName: String): Int
 
-	def lookup(code: Int) : String
+  def lookup(code: Int): String
 
 }

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/channel/ProtoBufCodec.scala

@@ -1,5 +1,12 @@
 package com.twitter.finagle.protobuf.rpc.channel
 
+import org.jboss.netty.channel.ChannelPipelineFactory
+import org.jboss.netty.channel.Channels
+import com.google.protobuf.Message
+import com.google.protobuf.Service
+import com.twitter.conversions.storage.intToStorageUnitableWholeNumber
+import com.twitter.finagle.Codec
+import com.twitter.finagle.CodecFactory
 
 class ProtoBufCodec(val service: Service) extends CodecFactory[(String, Message), (String, Message)] {
 

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/channel/ProtobufDecoder.scala

@@ -1,6 +1,11 @@
 package com.twitter.finagle.protobuf.rpc.channel
 
+import org.jboss.netty.buffer.ChannelBuffer
+import org.jboss.netty.buffer.ChannelBufferInputStream
+import org.jboss.netty.channel.Channel
+import org.jboss.netty.channel.ChannelHandlerContext
 
+import com.google.protobuf.Message
 /**
  *
  * Knows how to decode a ("doSomething()", ProtobufMessage) binary message.
@@ -9,9 +14,9 @@ package com.twitter.finagle.protobuf.rpc.channel
  * ==============
  *
  * Offset: 0             4                8
- *         +-------------+----------------+------------------+
- *         | method code | message length | protobuf message |
- *         +-------------+----------------+------------------+
+ * +-------------+----------------+------------------+
+ * | method code | message length | protobuf message |
+ * +-------------+----------------+------------------+
  *
  */
 

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/channel/ProtobufEncoder.scala

@@ -1,5 +1,11 @@
 package com.twitter.finagle.protobuf.rpc.channel
 
+import org.jboss.netty.handler.codec.oneone.OneToOneEncoder
+import org.jboss.netty.channel.ChannelHandlerContext
+import org.jboss.netty.channel.Channel
+import org.jboss.netty.buffer.ChannelBuffers
+
+import com.google.protobuf.Message
 
 /**
  * Knows how to encode a ("doSomething()", ProtobufMessage) tuple.

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/channel/ServerSideDecoder.scala

@@ -1,5 +1,12 @@
 package com.twitter.finagle.protobuf.rpc.channel
 
+import org.jboss.netty.buffer.ChannelBuffer
+import org.jboss.netty.channel.Channel
+import org.jboss.netty.channel.ChannelHandlerContext
+import org.jboss.netty.handler.codec.frame.FrameDecoder
+
+import com.google.protobuf.Message
+import com.google.protobuf.Service
 
 class ServerSideDecoder(val repo: MethodLookup, val service: Service) extends FrameDecoder with ProtobufDecoder {
 

finagle-protobuf/src/main/scala/com/twitter/finagle/protobuf/rpc/channel/SimpleMethodLookup.scala

@@ -1,27 +1,40 @@
 package com.twitter.finagle.protobuf.rpc.channel
 
 import com.twitter.finagle.protobuf.rpc.Util
+import com.google.protobuf.Service
+import scala.collection.JavaConversions._
+import scala.collection.mutable._
+import com.google.common.base.Joiner
 
 /**
  * Maps method names to short codes. This particular implementation uses the trivial String.hashCode.
  */
 class SimpleMethodLookup(val methods: java.util.List[String]) extends MethodLookup {
 
   val codeToNameMulti = new HashMap[Int, Set[String]] with MultiMap[Int, String]
-  methods.toList foreach { it => codeToNameMulti.getOrElseUpdate(createEncoding(it), Set.empty[String]) += it }
+  methods.toList foreach {
+    it => codeToNameMulti.getOrElseUpdate(createEncoding(it), Set.empty[String]) += it
+  }
 
   // do we have collisions?
-  codeToNameMulti foreach { it =>
-    if (it._2.size > 1) {
-      val sb = new StringBuilder("Collision on ")
-      it._2 foreach { it => sb.append(it).append(", ") }
-      throw new IllegalArgumentException(sb.toString())
-    }
+  codeToNameMulti foreach {
+    it =>
+      if (it._2.size > 1) {
+        val sb = new StringBuilder("Collision on ")
+        it._2 foreach {
+          it => sb.append(it).append(", ")
+        }
+        throw new IllegalArgumentException(sb.toString())
+      }
   }
 
   val codeToName = new HashMap[Int, String]
-  codeToNameMulti foreach { it => codeToName(it._1) = it._2.iterator.next() }
-  val nameToCode = codeToName map { _.swap }
+  codeToNameMulti foreach {
+    it => codeToName(it._1) = it._2.iterator.next()
+  }
+  val nameToCode = codeToName map {
+    _.swap
+  }
 
   def encode(methodName: String): Int = nameToCode.getOrElse(methodName, throw new NoSuchMethodException("Method: " + methodName))
 
@@ -34,6 +47,8 @@ class SimpleMethodLookup(val methods: java.util.List[String]) extends MethodLook
 }
 
 object SimpleMethodLookup {
-  def apply(s: Service): SimpleMethodLookup = { new SimpleMethodLookup(Util.extractMethodNames(s)) }
+  def apply(s: Service): SimpleMethodLookup = {
+    new SimpleMethodLookup(Util.extractMethodNames(s))
+  }
 }