-
-
Notifications
You must be signed in to change notification settings - Fork 281
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use scalapb-runtime-grpc to avoid conversions to/from Java.
- Loading branch information
Showing
6 changed files
with
152 additions
and
50 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
16 changes: 16 additions & 0 deletions
16
scalapb-runtime-grpc/src/main/scala/com/trueaccord/scalapb/grpc/Grpc.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package com.trueaccord.scalapb.grpc | ||
|
|
||
| import com.google.common.util.concurrent.{FutureCallback, Futures, ListenableFuture} | ||
|
|
||
| import scala.concurrent.{Promise, Future} | ||
|
|
||
| object Grpc { | ||
| def guavaFuture2ScalaFuture[A](guavaFuture: ListenableFuture[A]): Future[A] = { | ||
| val p = Promise[A]() | ||
| Futures.addCallback(guavaFuture, new FutureCallback[A] { | ||
| override def onFailure(t: Throwable) = p.failure(t) | ||
| override def onSuccess(a: A) = p.success(a) | ||
| }) | ||
| p.future | ||
| } | ||
| } |
18 changes: 18 additions & 0 deletions
18
scalapb-runtime-grpc/src/main/scala/com/trueaccord/scalapb/grpc/Marshaller.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| package com.trueaccord.scalapb.grpc | ||
|
|
||
| import java.io.InputStream | ||
|
|
||
| import com.trueaccord.scalapb.grpc.ProtoInputStream.NotStarted | ||
| import com.trueaccord.scalapb.{GeneratedMessage, GeneratedMessageCompanion, Message} | ||
|
|
||
| class Marshaller[T <: GeneratedMessage with Message[T]](companion: GeneratedMessageCompanion[T]) extends io.grpc.MethodDescriptor.Marshaller[T] { | ||
| override def stream(t: T): InputStream = ProtoInputStream.newInstance(t) | ||
|
|
||
| override def parse(inputStream: InputStream): T = inputStream match { | ||
| /* Optimization for in-memory transport. */ | ||
| case ProtoInputStream(NotStarted(m: T @unchecked)) if m.companion == companion => | ||
| m | ||
| case is => companion.parseFrom(is) | ||
| } | ||
| } | ||
|
|
83 changes: 83 additions & 0 deletions
83
scalapb-runtime-grpc/src/main/scala/com/trueaccord/scalapb/grpc/ProtoInputStream.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| package com.trueaccord.scalapb.grpc | ||
|
|
||
| import java.io.{ByteArrayInputStream, IOException, InputStream, OutputStream} | ||
|
|
||
| import com.google.common.io.ByteStreams | ||
| import com.google.protobuf.CodedOutputStream | ||
| import com.trueaccord.scalapb.GeneratedMessage | ||
| import com.trueaccord.scalapb.grpc.ProtoInputStream._ | ||
|
|
||
| private class ProtoInputStream(var state: State) extends InputStream { | ||
| @throws(classOf[IOException]) | ||
| def drainTo(target: OutputStream): Int = { | ||
| val bytesWritten = state match { | ||
| case NotStarted(message) => | ||
| message.writeTo(target) | ||
| message.serializedSize | ||
| case Partial(partial) => | ||
| ByteStreams.copy(partial, target).toInt | ||
| case Done => throw new IllegalStateException() | ||
| } | ||
| state = Done | ||
| bytesWritten | ||
| } | ||
|
|
||
| @throws(classOf[IOException]) | ||
| def read: Int = { | ||
| state match { | ||
| case NotStarted(message) => | ||
| state = Partial(new ByteArrayInputStream(message.toByteArray)) | ||
| case _ => | ||
| } | ||
| state match { | ||
| case Partial(partial) => | ||
| partial.read() | ||
| case _ => | ||
| -1 | ||
| } | ||
| } | ||
|
|
||
| @throws(classOf[IOException]) | ||
| override def read(b: Array[Byte], off: Int, len: Int): Int = | ||
| state match { | ||
| case NotStarted(message) => | ||
| message.serializedSize match { | ||
| case 0 => | ||
| state = Done | ||
| -1 | ||
| case size if len >= size => | ||
| val stream = CodedOutputStream.newInstance(b, off, size) | ||
| message.writeTo(stream) | ||
| stream.flush() | ||
| stream.checkNoSpaceLeft() | ||
| state = Done | ||
| size | ||
| case _ => | ||
| val partial = new ByteArrayInputStream(message.toByteArray) | ||
| state = Partial(partial) | ||
| partial.read(b, off, len) | ||
| } | ||
| case Partial(partial) => | ||
| partial.read(b, off, len) | ||
| case _ => | ||
| -1 | ||
| } | ||
|
|
||
| @throws(classOf[IOException]) | ||
| override def available: Int = state match { | ||
| case NotStarted(message) => message.serializedSize | ||
| case Partial(partial) => partial.available() | ||
| case _ => 0 | ||
| } | ||
| } | ||
|
|
||
| private object ProtoInputStream { | ||
| sealed trait State | ||
| case class NotStarted[A <: GeneratedMessage](message: A) extends State | ||
| case class Partial(bytes: ByteArrayInputStream) extends State | ||
| case object Done extends State | ||
|
|
||
| def unapply(p: ProtoInputStream): Option[State] = Some(p.state) | ||
|
|
||
| def newInstance(message: GeneratedMessage): InputStream = new ProtoInputStream(NotStarted(message)) | ||
| } |