-
Notifications
You must be signed in to change notification settings - Fork 3.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implementing Netty-based GRPC Client Transport. Server-side will be i…
…mplemented in a follow-up CL. Overview: - NettyClientTransport - this is the entry point into the client transport. This creates streams as well as the NettyClientHandler - NettyClientStream - client stream implementation. This sends commands to the NettyClientHandler via commands. Callbacks on the stream are made directly from the NettyClientHandler in the channel thread context. - NettyClientHandler - A Netty HTTP/2 handler that acts as a bridge between the NettyClientStreams and Netty's HTTP/2 processing. ------------- Created by MOE: http://code.google.com/p/moe-java MOE_MIGRATED_REVID=70970028
- Loading branch information
Showing
20 changed files
with
1,774 additions
and
110 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
8 changes: 3 additions & 5 deletions
8
core/src/main/java/com/google/net/stubby/newtransport/ClientStream.java
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 |
|---|---|---|
| @@ -1,16 +1,14 @@ | ||
| package com.google.net.stubby.newtransport; | ||
|
|
||
|
|
||
| /** | ||
| * Extension of {@link Stream} to support client-side termination semantics. | ||
| */ | ||
| public interface ClientStream extends Stream { | ||
|
|
||
| /** | ||
| * Used to abnormally terminate the stream. Any internally buffered messages are dropped. After | ||
| * this is called, no further messages may be sent and no further {@link StreamListener} callbacks | ||
| * (with the exception of onClosed) will be invoked for this stream. Any frames received for this | ||
| * stream after returning from this method will be discarded. | ||
| * Used to abnormally terminate the stream. After calling this method, no further messages will be | ||
| * sent or received, however it may still be possible to receive buffered messages for a brief | ||
| * period until {@link StreamListener#closed} is called. | ||
| */ | ||
| void cancel(); | ||
| } |
1 change: 1 addition & 0 deletions
1
core/src/main/java/com/google/net/stubby/newtransport/ClientTransportFactory.java
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
24 changes: 24 additions & 0 deletions
24
core/src/main/java/com/google/net/stubby/newtransport/HttpUtil.java
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,24 @@ | ||
| package com.google.net.stubby.newtransport; | ||
|
|
||
| /** | ||
| * Constants for GRPC-over-HTTP (or HTTP/2) | ||
| */ | ||
| public final class HttpUtil { | ||
| /** | ||
| * The Content-Type header name. Defined here since it is not explicitly defined by the HTTP/2 | ||
| * spec. | ||
| */ | ||
| public static final String CONTENT_TYPE_HEADER = "content-type"; | ||
|
|
||
| /** | ||
| * Content-Type used for GRPC-over-HTTP/2. | ||
| */ | ||
| public static final String CONTENT_TYPE_PROTORPC = "application/protorpc"; | ||
|
|
||
| /** | ||
| * The HTTP method used for GRPC requests. | ||
| */ | ||
| public static final String HTTP_METHOD = "POST"; | ||
|
|
||
| private HttpUtil() {} | ||
| } |
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
71 changes: 71 additions & 0 deletions
71
core/src/main/java/com/google/net/stubby/newtransport/netty/ByteBufDeframer.java
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,71 @@ | ||
| package com.google.net.stubby.newtransport.netty; | ||
|
|
||
| import com.google.net.stubby.newtransport.Deframer; | ||
| import com.google.net.stubby.newtransport.Framer; | ||
| import com.google.net.stubby.newtransport.TransportFrameUtil; | ||
|
|
||
| import io.netty.buffer.ByteBuf; | ||
| import io.netty.buffer.ByteBufAllocator; | ||
| import io.netty.buffer.ByteBufInputStream; | ||
| import io.netty.buffer.CompositeByteBuf; | ||
| import io.netty.buffer.UnpooledByteBufAllocator; | ||
|
|
||
| import java.io.DataInputStream; | ||
| import java.io.IOException; | ||
| import java.nio.ByteOrder; | ||
|
|
||
| /** | ||
| * Parse a sequence of {@link ByteBuf} instances that represent the frames of a GRPC call | ||
| */ | ||
| public class ByteBufDeframer extends Deframer<ByteBuf> { | ||
|
|
||
| private final CompositeByteBuf buffer; | ||
|
|
||
| public ByteBufDeframer(Framer target) { | ||
| this(UnpooledByteBufAllocator.DEFAULT, target); | ||
| } | ||
|
|
||
| public ByteBufDeframer(ByteBufAllocator alloc, Framer target) { | ||
| super(target); | ||
| buffer = alloc.compositeBuffer(); | ||
| } | ||
|
|
||
| public void dispose() { | ||
| // Remove the components from the composite buffer. This should set the reference | ||
| // count on all buffers to zero. | ||
| buffer.removeComponents(0, buffer.numComponents()); | ||
|
|
||
| // Release the composite buffer | ||
| buffer.release(); | ||
| } | ||
|
|
||
| @Override | ||
| protected DataInputStream prefix(ByteBuf frame) throws IOException { | ||
| buffer.addComponent(frame); | ||
| buffer.writerIndex(buffer.writerIndex() + frame.writerIndex() - frame.readerIndex()); | ||
| return new DataInputStream(new ByteBufInputStream(buffer)); | ||
| } | ||
|
|
||
| @Override | ||
| protected int consolidate() { | ||
| buffer.consolidate(); | ||
| return buffer.readableBytes(); | ||
| } | ||
|
|
||
| @Override | ||
| protected ByteBuf decompress(ByteBuf frame) throws IOException { | ||
| frame = frame.order(ByteOrder.BIG_ENDIAN); | ||
| int compressionType = frame.readUnsignedByte(); | ||
| int frameLength = frame.readUnsignedMedium(); | ||
| if (frameLength != frame.readableBytes()) { | ||
| throw new IllegalArgumentException("GRPC and buffer lengths misaligned. Frame length=" | ||
| + frameLength + ", readableBytes=" + frame.readableBytes()); | ||
| } | ||
| if (TransportFrameUtil.isNotCompressed(compressionType)) { | ||
| // Need to retain the frame as we may be holding it over channel events | ||
| frame.retain(); | ||
| return frame; | ||
| } | ||
| throw new IOException("Unknown compression type " + compressionType); | ||
| } | ||
| } |
18 changes: 18 additions & 0 deletions
18
core/src/main/java/com/google/net/stubby/newtransport/netty/CancelStreamCommand.java
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.google.net.stubby.newtransport.netty; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
|
|
||
| /** | ||
| * Command sent from a Netty client stream to the handler to cancel the stream. | ||
| */ | ||
| class CancelStreamCommand { | ||
| private final NettyClientStream stream; | ||
|
|
||
| CancelStreamCommand(NettyClientStream stream) { | ||
| this.stream = Preconditions.checkNotNull(stream, "stream"); | ||
| } | ||
|
|
||
| NettyClientStream stream() { | ||
| return stream; | ||
| } | ||
| } |
26 changes: 26 additions & 0 deletions
26
core/src/main/java/com/google/net/stubby/newtransport/netty/CreateStreamCommand.java
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,26 @@ | ||
| package com.google.net.stubby.newtransport.netty; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
| import com.google.net.stubby.MethodDescriptor; | ||
|
|
||
| /** | ||
| * A command to create a new stream. This is created by {@link NettyClientStream} and passed to the | ||
| * {@link NettyClientHandler} for processing in the Channel thread. | ||
| */ | ||
| class CreateStreamCommand { | ||
| final MethodDescriptor<?, ?> method; | ||
| final NettyClientStream stream; | ||
|
|
||
| CreateStreamCommand(MethodDescriptor<?, ?> method, NettyClientStream stream) { | ||
| this.method = Preconditions.checkNotNull(method, "method"); | ||
| this.stream = Preconditions.checkNotNull(stream, "stream"); | ||
| } | ||
|
|
||
| MethodDescriptor<?, ?> method() { | ||
| return method; | ||
| } | ||
|
|
||
| NettyClientStream stream() { | ||
| return stream; | ||
| } | ||
| } |
Oops, something went wrong.