grpc
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Updating gRPC code to the latest Netty version which supports binary headers. 

-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=75853353
This commit is contained in:
nathanmittler 2014-09-18 16:10:01 -07:00 committed by Eric Anderson
parent ef2129c7d2
commit fc7a052c07
13 changed files with 153 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
core/src/main/java/com/google/net/stubby/http2/netty/Http2Codec.java
View File

@ -15,6 +15,7 @@ import com.google.net.stubby.transport.Transport.Code;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http2.AbstractHttp2ConnectionHandler; import io.netty.handler.codec.http2.AbstractHttp2ConnectionHandler;
import io.netty.handler.codec.http2.DefaultHttp2Connection; import io.netty.handler.codec.http2.DefaultHttp2Connection;
import io.netty.handler.codec.http2.Http2Connection; import io.netty.handler.codec.http2.Http2Connection;
@ -23,12 +24,13 @@ import io.netty.handler.codec.http2.Http2Exception;
import io.netty.handler.codec.http2.Http2Headers; import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2Settings; import io.netty.handler.codec.http2.Http2Settings;
import java.util.Map;
/** /**
* Codec used by clients and servers to interpret HTTP2 frames in the context of an ongoing * Codec used by clients and servers to interpret HTTP2 frames in the context of an ongoing
* request-response dialog * request-response dialog
*/ */
public class Http2Codec extends AbstractHttp2ConnectionHandler { public class Http2Codec extends AbstractHttp2ConnectionHandler {
public static final int PADDING = 0; public static final int PADDING = 0;
private final RequestRegistry requestRegistry; private final RequestRegistry requestRegistry;
private final Session session; private final Session session;
@ -201,16 +203,26 @@ public class Http2Codec extends AbstractHttp2ConnectionHandler {
* Start the Request operation on the server * Start the Request operation on the server
*/ */
private Request serverStart(ChannelHandlerContext ctx, int streamId, Http2Headers headers) { private Request serverStart(ChannelHandlerContext ctx, int streamId, Http2Headers headers) {
if (!Http2Session.PROTORPC.equals(headers.get("content-type"))) { if (!Http2Session.PROTORPC.equals(headers.get(Http2Session.CONTENT_TYPE))) {
return null; return null;
} }
// Use Path to specify the operation // Use Path to specify the operation
String operationName = String operationName =
normalizeOperationName(headers.get(Http2Headers.PseudoHeaderName.PATH.value())); normalizeOperationName(headers.get(Http2Headers.PseudoHeaderName.PATH.value()).toString());
if (operationName == null) { if (operationName == null) {
return null; return null;
} }
Metadata.Headers grpcHeaders = new Metadata.Headers(headers);
// The Netty AsciiString class is really just a wrapper around a byte[] and supports
// arbitrary binary data, not just ASCII.
byte[][] headerValues = new byte[headers.size() * 2][];
int i = 0;
for (Map.Entry<AsciiString, AsciiString> entry : headers) {
headerValues[i++] = entry.getKey().array();
headerValues[i++] = entry.getValue().array();
}
Metadata.Headers grpcHeaders = new Metadata.Headers(headerValues);
// Create the operation and bind a HTTP2 response operation // Create the operation and bind a HTTP2 response operation
Request op = session.startRequest(operationName, grpcHeaders, Request op = session.startRequest(operationName, grpcHeaders,
createResponse(new Http2Writer(ctx), streamId)); createResponse(new Http2Writer(ctx), streamId));

28
core/src/main/java/com/google/net/stubby/http2/netty/Http2Request.java
View File

@ -5,7 +5,9 @@ import com.google.net.stubby.Request;
import com.google.net.stubby.Response; import com.google.net.stubby.Response;
import com.google.net.stubby.transport.Framer; import com.google.net.stubby.transport.Framer;
import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http2.DefaultHttp2Headers; import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.Http2Headers;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.UnknownHostException; import java.net.UnknownHostException;
@ -14,9 +16,10 @@ import java.net.UnknownHostException;
* A HTTP2 based implementation of {@link Request} * A HTTP2 based implementation of {@link Request}
*/ */
class Http2Request extends Http2Operation implements Request { class Http2Request extends Http2Operation implements Request {
private static final AsciiString POST = new AsciiString("POST");
private static final AsciiString HOST_NAME;
private static final AsciiString HTTPS = new AsciiString("https");
// TODO(user): Inject this // TODO(user): Inject this
private static final String HOST_NAME;
static { static {
String hostName; String hostName;
try { try {
@ -24,7 +27,7 @@ class Http2Request extends Http2Operation implements Request {
} catch (UnknownHostException uhe) { } catch (UnknownHostException uhe) {
hostName = "localhost"; hostName = "localhost";
} }
HOST_NAME = hostName; HOST_NAME = new AsciiString(hostName);
} }
private final Response response; private final Response response;
@ -33,19 +36,18 @@ class Http2Request extends Http2Operation implements Request {
Metadata.Headers headers, Metadata.Headers headers,
Http2Codec.Http2Writer writer, Framer framer) { Http2Codec.Http2Writer writer, Framer framer) {
super(response.getId(), writer, framer); super(response.getId(), writer, framer);
DefaultHttp2Headers.Builder headersBuilder = DefaultHttp2Headers.newBuilder(); Http2Headers http2Headers = new DefaultHttp2Headers();
// TODO(user) Switch the ASCII requirement to false once Netty supports binary byte[][] headerValues = headers.serialize();
// headers.
String[] headerValues = headers.serializeAscii();
for (int i = 0; i < headerValues.length; i++) { for (int i = 0; i < headerValues.length; i++) {
headersBuilder.add(headerValues[i], headerValues[++i]); http2Headers.add(new AsciiString(headerValues[i], false),
new AsciiString(headerValues[++i], false));
} }
headersBuilder.method("POST") http2Headers.method(POST)
.path("/" + operationName) .path(new AsciiString("/" + operationName))
.authority(HOST_NAME) .authority(HOST_NAME)
.scheme("https") .scheme(HTTPS)
.add("content-type", Http2Session.PROTORPC); .add(Http2Session.CONTENT_TYPE, Http2Session.PROTORPC);
writer.writeHeaders(response.getId(), headersBuilder.build(), false); writer.writeHeaders(response.getId(), http2Headers, false);
this.response = response; this.response = response;
} }

8
core/src/main/java/com/google/net/stubby/http2/netty/Http2Response.java
View File

@ -3,14 +3,16 @@ package com.google.net.stubby.http2.netty;
import com.google.net.stubby.Response; import com.google.net.stubby.Response;
import com.google.net.stubby.transport.Framer; import com.google.net.stubby.transport.Framer;
import io.netty.handler.codec.http2.Http2Headers; import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.DefaultHttp2Headers; import io.netty.handler.codec.http2.DefaultHttp2Headers;
/** /**
* A HTTP2 based implementation of a {@link Response}. * A HTTP2 based implementation of a {@link Response}.
*/ */
class Http2Response extends Http2Operation implements Response { class Http2Response extends Http2Operation implements Response {
private static final AsciiString STATUS_OK = new AsciiString("200");
public static ResponseBuilder builder(final int id, final Http2Codec.Http2Writer writer, public static ResponseBuilder builder(final int id, final Http2Codec.Http2Writer writer,
final Framer framer) { final Framer framer) {
@ -29,8 +31,8 @@ class Http2Response extends Http2Operation implements Response {
private Http2Response(int id, Http2Codec.Http2Writer writer, Framer framer) { private Http2Response(int id, Http2Codec.Http2Writer writer, Framer framer) {
super(id, writer, framer); super(id, writer, framer);
Http2Headers headers = DefaultHttp2Headers.newBuilder().status("200") Http2Headers headers = new DefaultHttp2Headers().status(STATUS_OK)
.add("content-type", Http2Session.PROTORPC).build(); .add(Http2Session.CONTENT_TYPE, Http2Session.PROTORPC);
writer.writeHeaders(id, headers, false); writer.writeHeaders(id, headers, false);
} }
} }

5
core/src/main/java/com/google/net/stubby/http2/netty/Http2Session.java
View File

@ -7,6 +7,8 @@ import com.google.net.stubby.Response;
import com.google.net.stubby.Session; import com.google.net.stubby.Session;
import com.google.net.stubby.transport.MessageFramer; import com.google.net.stubby.transport.MessageFramer;
import io.netty.handler.codec.AsciiString;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
/** /**
@ -15,7 +17,8 @@ import java.util.concurrent.atomic.AtomicInteger;
*/ */
public class Http2Session implements Session { public class Http2Session implements Session {
public static final String PROTORPC = "application/protorpc"; public static final AsciiString CONTENT_TYPE = new AsciiString("content-type");
public static final AsciiString PROTORPC = new AsciiString("application/protorpc");
private final Http2Codec.Http2Writer writer; private final Http2Codec.Http2Writer writer;
private final RequestRegistry requestRegistry; private final RequestRegistry requestRegistry;

1
core/src/main/java/com/google/net/stubby/newtransport/ServerTransportListener.java
View File

@ -1,7 +1,6 @@
package com.google.net.stubby.newtransport; package com.google.net.stubby.newtransport;
import com.google.net.stubby.Metadata; import com.google.net.stubby.Metadata;
import com.google.net.stubby.MethodDescriptor;
/** /**
* A observer of a server-side transport for stream creation events. * A observer of a server-side transport for stream creation events.

9
core/src/main/java/com/google/net/stubby/newtransport/netty/NettyClientStream.java
View File

@ -9,13 +9,14 @@ import com.google.net.stubby.Metadata;
import com.google.net.stubby.Status; import com.google.net.stubby.Status;
import com.google.net.stubby.newtransport.AbstractClientStream; import com.google.net.stubby.newtransport.AbstractClientStream;
import com.google.net.stubby.newtransport.GrpcDeframer; import com.google.net.stubby.newtransport.GrpcDeframer;
import com.google.net.stubby.newtransport.MessageDeframer2;
import com.google.net.stubby.newtransport.HttpUtil; import com.google.net.stubby.newtransport.HttpUtil;
import com.google.net.stubby.newtransport.MessageDeframer2;
import com.google.net.stubby.newtransport.StreamListener; import com.google.net.stubby.newtransport.StreamListener;
import com.google.net.stubby.transport.Transport; import com.google.net.stubby.transport.Transport;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http2.DefaultHttp2InboundFlowController; import io.netty.handler.codec.http2.DefaultHttp2InboundFlowController;
import io.netty.handler.codec.http2.Http2Headers; import io.netty.handler.codec.http2.Http2Headers;
@ -148,8 +149,8 @@ class NettyClientStream extends AbstractClientStream implements NettyStream {
return true; return true;
} }
String contentType = headers.get(HttpUtil.CONTENT_TYPE_HEADER); AsciiString contentType = headers.get(Utils.CONTENT_TYPE_HEADER);
return HttpUtil.CONTENT_TYPE_PROTORPC.equalsIgnoreCase(contentType); return Utils.CONTENT_TYPE_PROTORPC.equalsIgnoreCase(contentType);
} }
/** /**
@ -160,7 +161,7 @@ class NettyClientStream extends AbstractClientStream implements NettyStream {
return Transport.Code.UNKNOWN; return Transport.Code.UNKNOWN;
} }
String statusLine = headers.status(); AsciiString statusLine = headers.status();
if (statusLine == null) { if (statusLine == null) {
return Transport.Code.UNKNOWN; return Transport.Code.UNKNOWN;
} }

7
core/src/main/java/com/google/net/stubby/newtransport/netty/NettyClientTransport.java
View File

@ -19,6 +19,7 @@ import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup; import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http2.DefaultHttp2Connection; import io.netty.handler.codec.http2.DefaultHttp2Connection;
import io.netty.handler.codec.http2.DefaultHttp2FrameReader; import io.netty.handler.codec.http2.DefaultHttp2FrameReader;
import io.netty.handler.codec.http2.DefaultHttp2FrameWriter; import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
@ -50,7 +51,7 @@ class NettyClientTransport extends AbstractClientTransport {
private final Http2Negotiator.Negotiation negotiation; private final Http2Negotiator.Negotiation negotiation;
private final NettyClientHandler handler; private final NettyClientHandler handler;
private final boolean ssl; private final boolean ssl;
private final String authority; private final AsciiString authority;
private Channel channel; private Channel channel;
NettyClientTransport(InetSocketAddress address, NegotiationType negotiationType) { NettyClientTransport(InetSocketAddress address, NegotiationType negotiationType) {
@ -63,7 +64,7 @@ class NettyClientTransport extends AbstractClientTransport {
this.address = Preconditions.checkNotNull(address, "address"); this.address = Preconditions.checkNotNull(address, "address");
this.eventGroup = Preconditions.checkNotNull(eventGroup, "eventGroup"); this.eventGroup = Preconditions.checkNotNull(eventGroup, "eventGroup");
authority = address.getHostString() + ":" + address.getPort(); authority = new AsciiString(address.getHostString() + ":" + address.getPort());
handler = newHandler(); handler = newHandler();
switch (negotiationType) { switch (negotiationType) {
@ -94,7 +95,7 @@ class NettyClientTransport extends AbstractClientTransport {
try { try {
// Convert the headers into Netty HTTP/2 headers. // Convert the headers into Netty HTTP/2 headers.
String defaultPath = "/" + method.getName(); AsciiString defaultPath = new AsciiString("/" + method.getName());
Http2Headers http2Headers = Utils.convertHeaders(headers, ssl, defaultPath, authority); Http2Headers http2Headers = Utils.convertHeaders(headers, ssl, defaultPath, authority);
// Write the request and await creation of the stream. // Write the request and await creation of the stream.

28
core/src/main/java/com/google/net/stubby/newtransport/netty/NettyServerHandler.java
View File

@ -1,11 +1,11 @@
package com.google.net.stubby.newtransport.netty; package com.google.net.stubby.newtransport.netty;
import static com.google.net.stubby.newtransport.HttpUtil.CONTENT_TYPE_HEADER; import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_HEADER;
import static com.google.net.stubby.newtransport.HttpUtil.CONTENT_TYPE_PROTORPC; import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_PROTORPC;
import static com.google.net.stubby.newtransport.HttpUtil.HTTP_METHOD; import static com.google.net.stubby.newtransport.netty.Utils.HTTP_METHOD;
import static com.google.net.stubby.newtransport.netty.Utils.STATUS_OK;
import com.google.common.base.Preconditions; import com.google.common.base.Preconditions;
import com.google.net.stubby.Metadata;
import com.google.net.stubby.Status; import com.google.net.stubby.Status;
import com.google.net.stubby.newtransport.ServerStreamListener; import com.google.net.stubby.newtransport.ServerStreamListener;
import com.google.net.stubby.newtransport.ServerTransportListener; import com.google.net.stubby.newtransport.ServerTransportListener;
@ -33,7 +33,6 @@ import io.netty.handler.codec.http2.Http2StreamException;
import io.netty.util.ReferenceCountUtil; import io.netty.util.ReferenceCountUtil;
import java.util.List; import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level; import java.util.logging.Level;
import java.util.logging.Logger; import java.util.logging.Logger;
@ -86,7 +85,7 @@ class NettyServerHandler extends AbstractHttp2ConnectionHandler {
http2Stream.data(stream); http2Stream.data(stream);
String method = determineMethod(streamId, headers); String method = determineMethod(streamId, headers);
ServerStreamListener listener = transportListener.streamCreated(stream, method, ServerStreamListener listener = transportListener.streamCreated(stream, method,
new Metadata.Headers(headers)); Utils.convertHeaders(headers));
stream.setListener(listener); stream.setListener(listener);
} catch (Http2Exception e) { } catch (Http2Exception e) {
throw e; throw e;
@ -181,13 +180,14 @@ class NettyServerHandler extends AbstractHttp2ConnectionHandler {
writeData(ctx, cmd.streamId(), cmd.content(), 0, cmd.endStream(), promise); writeData(ctx, cmd.streamId(), cmd.content(), 0, cmd.endStream(), promise);
} else if (msg instanceof SendResponseHeadersCommand) { } else if (msg instanceof SendResponseHeadersCommand) {
SendResponseHeadersCommand cmd = (SendResponseHeadersCommand) msg; SendResponseHeadersCommand cmd = (SendResponseHeadersCommand) msg;
writeHeaders( writeHeaders(ctx,
ctx, cmd.streamId(), cmd.streamId(),
DefaultHttp2Headers.newBuilder() new DefaultHttp2Headers()
.status("200") .status(STATUS_OK)
.set(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC) .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC),
.build(), 0,
0, false, promise); false,
promise);
} else { } else {
AssertionError e = new AssertionError("Write called for unexpected type: " AssertionError e = new AssertionError("Write called for unexpected type: "
+ msg.getClass().getName()); + msg.getClass().getName());
@ -208,7 +208,7 @@ class NettyServerHandler extends AbstractHttp2ConnectionHandler {
String.format("Header '%s'='%s', while '%s' is expected", CONTENT_TYPE_HEADER, String.format("Header '%s'='%s', while '%s' is expected", CONTENT_TYPE_HEADER,
headers.get(CONTENT_TYPE_HEADER), CONTENT_TYPE_PROTORPC)); headers.get(CONTENT_TYPE_HEADER), CONTENT_TYPE_PROTORPC));
} }
String methodName = TransportFrameUtil.getFullMethodNameFromPath(headers.path()); String methodName = TransportFrameUtil.getFullMethodNameFromPath(headers.path().toString());
if (methodName == null) { if (methodName == null) {
throw new Http2StreamException(streamId, Http2Error.REFUSED_STREAM, throw new Http2StreamException(streamId, Http2Error.REFUSED_STREAM,
String.format("Malformatted path: %s", headers.path())); String.format("Malformatted path: %s", headers.path()));

70
core/src/main/java/com/google/net/stubby/newtransport/netty/Utils.java
View File

@ -1,29 +1,32 @@
package com.google.net.stubby.newtransport.netty; package com.google.net.stubby.newtransport.netty;
import static com.google.net.stubby.newtransport.HttpUtil.CONTENT_TYPE_HEADER;
import static com.google.net.stubby.newtransport.HttpUtil.CONTENT_TYPE_PROTORPC;
import static com.google.net.stubby.newtransport.HttpUtil.HTTP_METHOD;
import static io.netty.util.CharsetUtil.UTF_8;
import com.google.common.base.Preconditions; import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.net.stubby.Metadata; import com.google.net.stubby.Metadata;
import com.google.net.stubby.newtransport.HttpUtil;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.ByteBufAllocator;
import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http2.DefaultHttp2Headers; import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.Http2Headers; import io.netty.handler.codec.http2.Http2Headers;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.Map; import java.util.Map;
import javax.inject.Provider;
/** /**
* Common utility methods. * Common utility methods.
*/ */
class Utils { class Utils {
public static final AsciiString STATUS_OK = new AsciiString("200");
public static final AsciiString HTTP_METHOD = new AsciiString(HttpUtil.HTTP_METHOD);
public static final AsciiString HTTPS = new AsciiString("https");
public static final AsciiString HTTP = new AsciiString("http");
public static final AsciiString CONTENT_TYPE_HEADER =
new AsciiString(HttpUtil.CONTENT_TYPE_HEADER);
public static final AsciiString CONTENT_TYPE_PROTORPC =
new AsciiString(HttpUtil.CONTENT_TYPE_PROTORPC);
/** /**
* Copies the content of the given {@link ByteBuffer} to a new {@link ByteBuf} instance. * Copies the content of the given {@link ByteBuffer} to a new {@link ByteBuf} instance.
*/ */
@ -33,59 +36,52 @@ class Utils {
return buf; return buf;
} }
public static Metadata.Headers convertHeaders(Http2Headers http2Headers) {
// The Netty AsciiString class is really just a wrapper around a byte[] and supports
// arbitrary binary data, not just ASCII.
byte[][] headerValues = new byte[http2Headers.size()*2][];
int i = 0;
for (Map.Entry<AsciiString, AsciiString> entry : http2Headers) {
headerValues[i++] = entry.getKey().array();
headerValues[i++] = entry.getValue().array();
}
return new Metadata.Headers(headerValues);
}
public static Http2Headers convertHeaders(Metadata.Headers headers, public static Http2Headers convertHeaders(Metadata.Headers headers,
boolean ssl, boolean ssl,
String defaultPath, AsciiString defaultPath,
String defaultAuthority) { AsciiString defaultAuthority) {
Preconditions.checkNotNull(headers, "headers"); Preconditions.checkNotNull(headers, "headers");
Preconditions.checkNotNull(defaultPath, "defaultPath"); Preconditions.checkNotNull(defaultPath, "defaultPath");
Preconditions.checkNotNull(defaultAuthority, "defaultAuthority"); Preconditions.checkNotNull(defaultAuthority, "defaultAuthority");
DefaultHttp2Headers.Builder headersBuilder = DefaultHttp2Headers.newBuilder(); Http2Headers http2Headers = new DefaultHttp2Headers();
// Add any application-provided headers first. // Add any application-provided headers first.
byte[][] serializedHeaders = headers.serialize(); byte[][] serializedHeaders = headers.serialize();
for (int i = 0; i < serializedHeaders.length; i++) { for (int i = 0; i < serializedHeaders.length; i++) {
String key = new String(serializedHeaders[i], UTF_8); http2Headers.add(new AsciiString(serializedHeaders[i], false),
String value = new String(serializedHeaders[++i], UTF_8); new AsciiString(serializedHeaders[++i], false));
headersBuilder.add(key, value);
} }
// Now set GRPC-specific default headers. // Now set GRPC-specific default headers.
headersBuilder http2Headers
.authority(defaultAuthority) .authority(defaultAuthority)
.path(defaultPath) .path(defaultPath)
.method(HTTP_METHOD) .method(HTTP_METHOD)
.scheme(ssl? "https" : "http") .scheme(ssl? HTTPS : HTTP)
.add(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC); .add(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC);
// Override the default authority and path if provided by the headers. // Override the default authority and path if provided by the headers.
if (headers.getAuthority() != null) { if (headers.getAuthority() != null) {
headersBuilder.authority(headers.getAuthority()); http2Headers.authority(new AsciiString(headers.getAuthority()));
} }
if (headers.getPath() != null) { if (headers.getPath() != null) {
headersBuilder.path(headers.getPath()); http2Headers.path(new AsciiString(headers.getPath()));
} }
return headersBuilder.build(); return http2Headers;
}
public static ImmutableMap<String, Provider<String>> convertHeaders(Http2Headers headers) {
ImmutableMap.Builder<String, Provider<String>> grpcHeaders =
new ImmutableMap.Builder<String, Provider<String>>();
for (Map.Entry<String, String> header : headers) {
if (!header.getKey().startsWith(":")) {
final String value = header.getValue();
// headers starting with ":" are reserved for HTTP/2 built-in headers
grpcHeaders.put(header.getKey(), new Provider<String>() {
@Override
public String get() {
return value;
}
});
}
}
return grpcHeaders.build();
} }
private Utils() { private Utils() {

40
core/src/test/java/com/google/net/stubby/newtransport/netty/NettyClientHandlerTest.java
View File

@ -1,8 +1,10 @@
package com.google.net.stubby.newtransport.netty; package com.google.net.stubby.newtransport.netty;
import static com.google.net.stubby.newtransport.HttpUtil.CONTENT_TYPE_HEADER; import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_HEADER;
import static com.google.net.stubby.newtransport.HttpUtil.CONTENT_TYPE_PROTORPC; import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_PROTORPC;
import static com.google.net.stubby.newtransport.HttpUtil.HTTP_METHOD; import static com.google.net.stubby.newtransport.netty.Utils.HTTPS;
import static com.google.net.stubby.newtransport.netty.Utils.HTTP_METHOD;
import static com.google.net.stubby.newtransport.netty.Utils.STATUS_OK;
import static java.nio.charset.StandardCharsets.UTF_8; import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.any; import static org.mockito.Matchers.any;
@ -16,13 +18,13 @@ import static org.mockito.Mockito.when;
import com.google.net.stubby.Metadata; import com.google.net.stubby.Metadata;
import com.google.net.stubby.Status; import com.google.net.stubby.Status;
import com.google.net.stubby.newtransport.HttpUtil;
import com.google.net.stubby.newtransport.StreamState; import com.google.net.stubby.newtransport.StreamState;
import com.google.net.stubby.transport.Transport; import com.google.net.stubby.transport.Transport;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled; import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http2.DefaultHttp2Connection; import io.netty.handler.codec.http2.DefaultHttp2Connection;
import io.netty.handler.codec.http2.DefaultHttp2FrameReader; import io.netty.handler.codec.http2.DefaultHttp2FrameReader;
import io.netty.handler.codec.http2.DefaultHttp2FrameWriter; import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
@ -76,15 +78,13 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase {
mockContext(); mockContext();
mockFuture(true); mockFuture(true);
grpcHeaders = DefaultHttp2Headers grpcHeaders = new DefaultHttp2Headers()
.newBuilder() .scheme(HTTPS)
.scheme("https") .authority(as("www.fake.com"))
.authority("www.fake.com") .path(as("/fakemethod"))
.path("/fakemethod")
.method(HTTP_METHOD) .method(HTTP_METHOD)
.add("auth", "sometoken") .add(as("auth"), as("sometoken"))
.add(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC) .add(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC);
.build();
when(stream.state()).thenReturn(StreamState.OPEN); when(stream.state()).thenReturn(StreamState.OPEN);
@ -121,12 +121,12 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase {
eq(0), eq(0),
eq(false)); eq(false));
Http2Headers headers = captor.getValue(); Http2Headers headers = captor.getValue();
assertEquals("https", headers.scheme()); assertEquals("https", headers.scheme().toString());
assertEquals(HTTP_METHOD, headers.method()); assertEquals(HTTP_METHOD, headers.method());
assertEquals("www.fake.com", headers.authority()); assertEquals("www.fake.com", headers.authority().toString());
assertEquals(CONTENT_TYPE_PROTORPC, headers.get(CONTENT_TYPE_HEADER)); assertEquals(CONTENT_TYPE_PROTORPC, headers.get(CONTENT_TYPE_HEADER));
assertEquals("/fakemethod", headers.path()); assertEquals("/fakemethod", headers.path().toString());
assertEquals("sometoken", headers.get("auth")); assertEquals("sometoken", headers.get(as("auth")).toString());
} }
@Test @Test
@ -170,8 +170,8 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase {
createStream(); createStream();
// Read a headers frame first. // Read a headers frame first.
Http2Headers headers = DefaultHttp2Headers.newBuilder().status("200") Http2Headers headers = new DefaultHttp2Headers().status(STATUS_OK)
.set(HttpUtil.CONTENT_TYPE_HEADER, HttpUtil.CONTENT_TYPE_PROTORPC).build(); .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC);
ByteBuf headersFrame = headersFrame(3, headers); ByteBuf headersFrame = headersFrame(3, headers);
handler.channelRead(this.ctx, headersFrame); handler.channelRead(this.ctx, headersFrame);
verify(stream).inboundHeadersRecieved(headers, false); verify(stream).inboundHeadersRecieved(headers, false);
@ -269,4 +269,8 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase {
inboundFlow, inboundFlow,
outboundFlow); outboundFlow);
} }
private AsciiString as(String string) {
return new AsciiString(string);
}
} }

9
core/src/test/java/com/google/net/stubby/newtransport/netty/NettyClientStreamTest.java
View File

@ -1,5 +1,8 @@
package com.google.net.stubby.newtransport.netty; package com.google.net.stubby.newtransport.netty;
import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_HEADER;
import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_PROTORPC;
import static com.google.net.stubby.newtransport.netty.Utils.STATUS_OK;
import static io.netty.util.CharsetUtil.UTF_8; import static io.netty.util.CharsetUtil.UTF_8;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.any; import static org.mockito.Matchers.any;
@ -9,7 +12,6 @@ import static org.mockito.Mockito.verify;
import com.google.net.stubby.Metadata; import com.google.net.stubby.Metadata;
import com.google.net.stubby.Status; import com.google.net.stubby.Status;
import com.google.net.stubby.newtransport.HttpUtil;
import com.google.net.stubby.newtransport.StreamState; import com.google.net.stubby.newtransport.StreamState;
import com.google.net.stubby.transport.Transport; import com.google.net.stubby.transport.Transport;
@ -131,7 +133,8 @@ public class NettyClientStreamTest extends NettyStreamTestBase {
} }
private Http2Headers grpcResponseHeaders() { private Http2Headers grpcResponseHeaders() {
return DefaultHttp2Headers.newBuilder().status("200") return new DefaultHttp2Headers()
.set(HttpUtil.CONTENT_TYPE_HEADER, HttpUtil.CONTENT_TYPE_PROTORPC).build(); .status(STATUS_OK)
.set(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC);
} }
} }

15
core/src/test/java/com/google/net/stubby/newtransport/netty/NettyServerHandlerTest.java
View File

@ -1,5 +1,8 @@
package com.google.net.stubby.newtransport.netty; package com.google.net.stubby.newtransport.netty;
import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_HEADER;
import static com.google.net.stubby.newtransport.netty.Utils.CONTENT_TYPE_PROTORPC;
import static com.google.net.stubby.newtransport.netty.Utils.HTTP_METHOD;
import static java.nio.charset.StandardCharsets.UTF_8; import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
@ -14,10 +17,8 @@ import static org.mockito.Mockito.when;
import com.google.common.io.ByteStreams; import com.google.common.io.ByteStreams;
import com.google.net.stubby.Metadata; import com.google.net.stubby.Metadata;
import com.google.net.stubby.MethodDescriptor;
import com.google.net.stubby.Status; import com.google.net.stubby.Status;
import com.google.net.stubby.newtransport.Framer; import com.google.net.stubby.newtransport.Framer;
import com.google.net.stubby.newtransport.HttpUtil;
import com.google.net.stubby.newtransport.MessageFramer; import com.google.net.stubby.newtransport.MessageFramer;
import com.google.net.stubby.newtransport.ServerStream; import com.google.net.stubby.newtransport.ServerStream;
import com.google.net.stubby.newtransport.ServerStreamListener; import com.google.net.stubby.newtransport.ServerStreamListener;
@ -27,6 +28,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled; import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.AsciiString;
import io.netty.handler.codec.http2.DefaultHttp2Connection; import io.netty.handler.codec.http2.DefaultHttp2Connection;
import io.netty.handler.codec.http2.DefaultHttp2FrameReader; import io.netty.handler.codec.http2.DefaultHttp2FrameReader;
import io.netty.handler.codec.http2.DefaultHttp2FrameWriter; import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
@ -162,11 +164,10 @@ public class NettyServerHandlerTest extends NettyHandlerTestBase {
} }
private void createStream() throws Exception { private void createStream() throws Exception {
Http2Headers headers = DefaultHttp2Headers.newBuilder() Http2Headers headers = new DefaultHttp2Headers()
.method(HttpUtil.HTTP_METHOD) .method(HTTP_METHOD)
.set(HttpUtil.CONTENT_TYPE_HEADER, HttpUtil.CONTENT_TYPE_PROTORPC) .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_PROTORPC)
.path("/foo.bar") .path(new AsciiString("/foo.bar"));
.build();
ByteBuf headersFrame = headersFrame(STREAM_ID, headers); ByteBuf headersFrame = headersFrame(STREAM_ID, headers);
handler.channelRead(ctx, headersFrame); handler.channelRead(ctx, headersFrame);
ArgumentCaptor<NettyServerStream> streamCaptor = ArgumentCaptor<NettyServerStream> streamCaptor =

43
core/src/test/java/com/google/net/stubby/newtransport/netty/NettyServerStreamTest.java
View File

@ -2,7 +2,6 @@ package com.google.net.stubby.newtransport.netty;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail; import static org.junit.Assert.fail;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.notNull; import static org.mockito.Matchers.notNull;
import static org.mockito.Matchers.same; import static org.mockito.Matchers.same;
import static org.mockito.Mockito.never; import static org.mockito.Mockito.never;
@ -26,7 +25,7 @@ import org.mockito.Mock;
public class NettyServerStreamTest extends NettyStreamTestBase { public class NettyServerStreamTest extends NettyStreamTestBase {
@Mock @Mock
protected ServerStreamListener listener; protected ServerStreamListener serverListener;
private Metadata.Trailers trailers = new Metadata.Trailers(); private Metadata.Trailers trailers = new Metadata.Trailers();
@Test @Test
@ -46,7 +45,7 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
} catch (IllegalStateException expected) { } catch (IllegalStateException expected) {
} }
assertEquals(StreamState.OPEN, stream.state()); assertEquals(StreamState.OPEN, stream.state());
verifyZeroInteractions(listener); verifyZeroInteractions(serverListener);
} }
@Test @Test
@ -55,12 +54,12 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
stream().close(Status.CANCELLED, trailers); stream().close(Status.CANCELLED, trailers);
verify(channel).writeAndFlush( verify(channel).writeAndFlush(
new SendGrpcFrameCommand(STREAM_ID, statusFrame(Status.CANCELLED), true)); new SendGrpcFrameCommand(STREAM_ID, statusFrame(Status.CANCELLED), true));
verifyZeroInteractions(listener); verifyZeroInteractions(serverListener);
// Sending complete. Listener gets closed() // Sending complete. Listener gets closed()
stream().complete(); stream().complete();
verify(listener).closed(Status.CANCELLED, trailers); verify(serverListener).closed(Status.CANCELLED, trailers);
assertEquals(StreamState.CLOSED, stream.state()); assertEquals(StreamState.CLOSED, stream.state());
verifyZeroInteractions(listener); verifyZeroInteractions(serverListener);
} }
@Test @Test
@ -68,17 +67,17 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
// Client half-closes. Listener gets halfClosed() // Client half-closes. Listener gets halfClosed()
stream().remoteEndClosed(); stream().remoteEndClosed();
assertEquals(StreamState.WRITE_ONLY, stream.state()); assertEquals(StreamState.WRITE_ONLY, stream.state());
verify(listener).halfClosed(); verify(serverListener).halfClosed();
// Server closes. Status sent // Server closes. Status sent
stream().close(Status.OK, trailers); stream().close(Status.OK, trailers);
verifyNoMoreInteractions(listener); verifyNoMoreInteractions(serverListener);
assertEquals(StreamState.CLOSED, stream.state()); assertEquals(StreamState.CLOSED, stream.state());
verify(channel).writeAndFlush( verify(channel).writeAndFlush(
new SendGrpcFrameCommand(STREAM_ID, statusFrame(Status.OK), true)); new SendGrpcFrameCommand(STREAM_ID, statusFrame(Status.OK), true));
// Sending and receiving complete. Listener gets closed() // Sending and receiving complete. Listener gets closed()
stream().complete(); stream().complete();
verify(listener).closed(Status.OK, trailers); verify(serverListener).closed(Status.OK, trailers);
verifyNoMoreInteractions(listener); verifyNoMoreInteractions(serverListener);
} }
@Test @Test
@ -86,7 +85,7 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
// Client half-closes. Listener gets halfClosed() // Client half-closes. Listener gets halfClosed()
stream().remoteEndClosed(); stream().remoteEndClosed();
assertEquals(StreamState.WRITE_ONLY, stream.state()); assertEquals(StreamState.WRITE_ONLY, stream.state());
verify(listener).halfClosed(); verify(serverListener).halfClosed();
// Client half-closes again. // Client half-closes again.
try { try {
stream().remoteEndClosed(); stream().remoteEndClosed();
@ -94,7 +93,7 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
} catch (IllegalStateException expected) { } catch (IllegalStateException expected) {
} }
assertEquals(StreamState.WRITE_ONLY, stream.state()); assertEquals(StreamState.WRITE_ONLY, stream.state());
verifyNoMoreInteractions(listener); verifyNoMoreInteractions(serverListener);
} }
@Test @Test
@ -102,9 +101,9 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
Status status = new Status(Transport.Code.INTERNAL, new Throwable()); Status status = new Status(Transport.Code.INTERNAL, new Throwable());
stream().abortStream(status, true); stream().abortStream(status, true);
assertEquals(StreamState.CLOSED, stream.state()); assertEquals(StreamState.CLOSED, stream.state());
verify(listener).closed(same(status), notNull(Metadata.Trailers.class)); verify(serverListener).closed(same(status), notNull(Metadata.Trailers.class));
verify(channel).writeAndFlush(new SendGrpcFrameCommand(STREAM_ID, statusFrame(status), true)); verify(channel).writeAndFlush(new SendGrpcFrameCommand(STREAM_ID, statusFrame(status), true));
verifyNoMoreInteractions(listener); verifyNoMoreInteractions(serverListener);
} }
@Test @Test
@ -112,10 +111,10 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
Status status = new Status(Transport.Code.INTERNAL, new Throwable()); Status status = new Status(Transport.Code.INTERNAL, new Throwable());
stream().abortStream(status, false); stream().abortStream(status, false);
assertEquals(StreamState.CLOSED, stream.state()); assertEquals(StreamState.CLOSED, stream.state());
verify(listener).closed(same(status), notNull(Metadata.Trailers.class)); verify(serverListener).closed(same(status), notNull(Metadata.Trailers.class));
verify(channel, never()).writeAndFlush( verify(channel, never()).writeAndFlush(
new SendGrpcFrameCommand(STREAM_ID, statusFrame(status), true)); new SendGrpcFrameCommand(STREAM_ID, statusFrame(status), true));
verifyNoMoreInteractions(listener); verifyNoMoreInteractions(serverListener);
} }
@Test @Test
@ -124,26 +123,26 @@ public class NettyServerStreamTest extends NettyStreamTestBase {
// Client half-closes. Listener gets halfClosed() // Client half-closes. Listener gets halfClosed()
stream().remoteEndClosed(); stream().remoteEndClosed();
assertEquals(StreamState.WRITE_ONLY, stream.state()); assertEquals(StreamState.WRITE_ONLY, stream.state());
verify(listener).halfClosed(); verify(serverListener).halfClosed();
// Abort // Abort
stream().abortStream(status, true); stream().abortStream(status, true);
verify(listener).closed(same(status), notNull(Metadata.Trailers.class)); verify(serverListener).closed(same(status), notNull(Metadata.Trailers.class));
assertEquals(StreamState.CLOSED, stream.state()); assertEquals(StreamState.CLOSED, stream.state());
verifyNoMoreInteractions(listener); verifyNoMoreInteractions(serverListener);
} }
@Override @Override
protected NettyStream createStream() { protected NettyStream createStream() {
NettyServerStream stream = new NettyServerStream(channel, STREAM_ID, inboundFlow); NettyServerStream stream = new NettyServerStream(channel, STREAM_ID, inboundFlow);
stream.setListener(listener); stream.setListener(serverListener);
assertEquals(StreamState.OPEN, stream.state()); assertEquals(StreamState.OPEN, stream.state());
verifyZeroInteractions(listener); verifyZeroInteractions(serverListener);
return stream; return stream;
} }
@Override @Override
protected ServerStreamListener listener() { protected ServerStreamListener listener() {
return listener; return serverListener;
} }
private NettyServerStream stream() { private NettyServerStream stream() {