simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 1 | package com.google.net.stubby.newtransport.okhttp; |
| 2 | |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 3 | import com.google.common.annotations.VisibleForTesting; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 4 | import com.google.common.base.Preconditions; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 5 | import com.google.common.io.ByteStreams; |
nathanmittler | 43d2fcc | 2014-08-29 21:58:54 -0700 | [diff] [blame] | 6 | import com.google.common.util.concurrent.ListenableFuture; |
lryan | e4bd1c7 | 2014-09-08 14:03:35 -0700 | [diff] [blame] | 7 | import com.google.net.stubby.Metadata; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 8 | import com.google.net.stubby.MethodDescriptor; |
| 9 | import com.google.net.stubby.Status; |
zhangkun | 048649e | 2014-08-28 15:52:03 -0700 | [diff] [blame] | 10 | import com.google.net.stubby.newtransport.AbstractClientStream; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 11 | import com.google.net.stubby.newtransport.AbstractClientTransport; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 12 | import com.google.net.stubby.newtransport.ClientStream; |
ejona | 6fc356b | 2014-09-22 12:49:20 -0700 | [diff] [blame] | 13 | import com.google.net.stubby.newtransport.ClientStreamListener; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 14 | import com.google.net.stubby.newtransport.ClientTransport; |
| 15 | import com.google.net.stubby.newtransport.InputStreamDeframer; |
simonma | 1966d5b | 2014-08-08 14:51:01 -0700 | [diff] [blame] | 16 | import com.google.net.stubby.newtransport.StreamState; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 17 | |
| 18 | import com.squareup.okhttp.internal.spdy.ErrorCode; |
| 19 | import com.squareup.okhttp.internal.spdy.FrameReader; |
| 20 | import com.squareup.okhttp.internal.spdy.Header; |
| 21 | import com.squareup.okhttp.internal.spdy.HeadersMode; |
brettmorgan | 56f5ec3 | 2014-07-18 16:54:50 -0700 | [diff] [blame] | 22 | import com.squareup.okhttp.internal.spdy.Http20Draft12; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 23 | import com.squareup.okhttp.internal.spdy.Settings; |
| 24 | import com.squareup.okhttp.internal.spdy.Variant; |
| 25 | |
zhangkun | 048649e | 2014-08-28 15:52:03 -0700 | [diff] [blame] | 26 | import okio.Buffer; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 27 | import okio.BufferedSink; |
| 28 | import okio.BufferedSource; |
zhangkun | 048649e | 2014-08-28 15:52:03 -0700 | [diff] [blame] | 29 | import okio.ByteString; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 30 | import okio.Okio; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 31 | |
| 32 | import java.io.IOException; |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 33 | import java.net.InetSocketAddress; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 34 | import java.net.Socket; |
| 35 | import java.nio.ByteBuffer; |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 36 | import java.util.ArrayList; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 37 | import java.util.Collections; |
| 38 | import java.util.HashMap; |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 39 | import java.util.Iterator; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 40 | import java.util.List; |
| 41 | import java.util.Map; |
simonma | f2c4c85 | 2014-08-06 10:00:56 -0700 | [diff] [blame] | 42 | import java.util.concurrent.Executor; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 43 | |
| 44 | import javax.annotation.concurrent.GuardedBy; |
| 45 | |
| 46 | /** |
| 47 | * A okhttp-based {@link ClientTransport} implementation. |
| 48 | */ |
| 49 | public class OkHttpClientTransport extends AbstractClientTransport { |
| 50 | /** The default initial window size in HTTP/2 is 64 KiB for the stream and connection. */ |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 51 | @VisibleForTesting |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 52 | static final int DEFAULT_INITIAL_WINDOW_SIZE = 64 * 1024; |
| 53 | |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 54 | private static final Map<ErrorCode, Status> ERROR_CODE_TO_STATUS; |
| 55 | static { |
| 56 | Map<ErrorCode, Status> errorToStatus = new HashMap<ErrorCode, Status>(); |
| 57 | errorToStatus.put(ErrorCode.NO_ERROR, Status.OK); |
| 58 | errorToStatus.put(ErrorCode.PROTOCOL_ERROR, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 59 | Status.INTERNAL.withDescription("Protocol error")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 60 | errorToStatus.put(ErrorCode.INVALID_STREAM, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 61 | Status.INTERNAL.withDescription("Invalid stream")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 62 | errorToStatus.put(ErrorCode.UNSUPPORTED_VERSION, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 63 | Status.INTERNAL.withDescription("Unsupported version")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 64 | errorToStatus.put(ErrorCode.STREAM_IN_USE, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 65 | Status.INTERNAL.withDescription("Stream in use")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 66 | errorToStatus.put(ErrorCode.STREAM_ALREADY_CLOSED, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 67 | Status.INTERNAL.withDescription("Stream already closed")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 68 | errorToStatus.put(ErrorCode.INTERNAL_ERROR, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 69 | Status.INTERNAL.withDescription("Internal error")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 70 | errorToStatus.put(ErrorCode.FLOW_CONTROL_ERROR, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 71 | Status.INTERNAL.withDescription("Flow control error")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 72 | errorToStatus.put(ErrorCode.STREAM_CLOSED, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 73 | Status.INTERNAL.withDescription("Stream closed")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 74 | errorToStatus.put(ErrorCode.FRAME_TOO_LARGE, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 75 | Status.INTERNAL.withDescription("Frame too large")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 76 | errorToStatus.put(ErrorCode.REFUSED_STREAM, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 77 | Status.INTERNAL.withDescription("Refused stream")); |
| 78 | errorToStatus.put(ErrorCode.CANCEL, Status.CANCELLED.withDescription("Cancelled")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 79 | errorToStatus.put(ErrorCode.COMPRESSION_ERROR, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 80 | Status.INTERNAL.withDescription("Compression error")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 81 | errorToStatus.put(ErrorCode.INVALID_CREDENTIALS, |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 82 | Status.PERMISSION_DENIED.withDescription("Invalid credentials")); |
simonma | 8f3e9ee | 2014-09-05 11:02:52 -0700 | [diff] [blame] | 83 | ERROR_CODE_TO_STATUS = Collections.unmodifiableMap(errorToStatus); |
| 84 | } |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 85 | |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 86 | private final InetSocketAddress address; |
| 87 | private final String defaultAuthority; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 88 | private FrameReader frameReader; |
| 89 | private AsyncFrameWriter frameWriter; |
lryan | e4bd1c7 | 2014-09-08 14:03:35 -0700 | [diff] [blame] | 90 | private final Object lock = new Object(); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 91 | @GuardedBy("lock") |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 92 | private int nextStreamId; |
| 93 | private final Map<Integer, OkHttpClientStream> streams = |
| 94 | Collections.synchronizedMap(new HashMap<Integer, OkHttpClientStream>()); |
simonma | f2c4c85 | 2014-08-06 10:00:56 -0700 | [diff] [blame] | 95 | private final Executor executor; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 96 | private int unacknowledgedBytesRead; |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 97 | private ClientFrameHandler clientFrameHandler; |
| 98 | // The status used to finish all active streams when the transport is closed. |
| 99 | @GuardedBy("lock") |
| 100 | private boolean goAway; |
| 101 | @GuardedBy("lock") |
| 102 | private Status goAwayStatus; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 103 | |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 104 | OkHttpClientTransport(InetSocketAddress address, Executor executor) { |
| 105 | this.address = Preconditions.checkNotNull(address); |
| 106 | defaultAuthority = address.getHostString() + ":" + address.getPort(); |
simonma | f2c4c85 | 2014-08-06 10:00:56 -0700 | [diff] [blame] | 107 | this.executor = Preconditions.checkNotNull(executor); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 108 | // Client initiated streams are odd, server initiated ones are even. Server should not need to |
| 109 | // use it. We start clients at 3 to avoid conflicting with HTTP negotiation. |
| 110 | nextStreamId = 3; |
| 111 | } |
| 112 | |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 113 | /** |
| 114 | * Create a transport connected to a fake peer for test. |
| 115 | */ |
| 116 | @VisibleForTesting |
simonma | f2c4c85 | 2014-08-06 10:00:56 -0700 | [diff] [blame] | 117 | OkHttpClientTransport(Executor executor, FrameReader frameReader, AsyncFrameWriter frameWriter, |
| 118 | int nextStreamId) { |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 119 | address = null; |
| 120 | defaultAuthority = "notarealauthority:80"; |
simonma | f2c4c85 | 2014-08-06 10:00:56 -0700 | [diff] [blame] | 121 | this.executor = Preconditions.checkNotNull(executor); |
| 122 | this.frameReader = Preconditions.checkNotNull(frameReader); |
| 123 | this.frameWriter = Preconditions.checkNotNull(frameWriter); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 124 | this.nextStreamId = nextStreamId; |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 125 | } |
| 126 | |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 127 | @Override |
lryan | e4bd1c7 | 2014-09-08 14:03:35 -0700 | [diff] [blame] | 128 | protected ClientStream newStreamInternal(MethodDescriptor<?, ?> method, |
| 129 | Metadata.Headers headers, |
ejona | 6fc356b | 2014-09-22 12:49:20 -0700 | [diff] [blame] | 130 | ClientStreamListener listener) { |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 131 | return new OkHttpClientStream(method, headers, listener); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 132 | } |
| 133 | |
| 134 | @Override |
| 135 | protected void doStart() { |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 136 | // We set host to null for test. |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 137 | if (address != null) { |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 138 | BufferedSource source; |
| 139 | BufferedSink sink; |
| 140 | try { |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 141 | Socket socket = new Socket(address.getAddress(), address.getPort()); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 142 | source = Okio.buffer(Okio.source(socket)); |
| 143 | sink = Okio.buffer(Okio.sink(socket)); |
| 144 | } catch (IOException e) { |
| 145 | throw new RuntimeException(e); |
| 146 | } |
| 147 | Variant variant = new Http20Draft12(); |
| 148 | frameReader = variant.newReader(source, true); |
| 149 | frameWriter = new AsyncFrameWriter(variant.newWriter(sink, true), this, executor); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 150 | } |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 151 | |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 152 | notifyStarted(); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 153 | clientFrameHandler = new ClientFrameHandler(); |
| 154 | executor.execute(clientFrameHandler); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 155 | } |
| 156 | |
| 157 | @Override |
| 158 | protected void doStop() { |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 159 | boolean normalClose; |
| 160 | synchronized (lock) { |
| 161 | normalClose = !goAway; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 162 | } |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 163 | if (normalClose) { |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 164 | abort(Status.INTERNAL.withDescription("Transport stopped")); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 165 | // Send GOAWAY with lastGoodStreamId of 0, since we don't expect any server-initiated streams. |
| 166 | // The GOAWAY is part of graceful shutdown. |
simonma | b645b38 | 2014-08-26 10:58:45 -0700 | [diff] [blame] | 167 | frameWriter.goAway(0, ErrorCode.NO_ERROR, new byte[0]); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 168 | } |
| 169 | stopIfNecessary(); |
| 170 | } |
| 171 | |
| 172 | @VisibleForTesting |
| 173 | ClientFrameHandler getHandler() { |
| 174 | return clientFrameHandler; |
| 175 | } |
| 176 | |
| 177 | @VisibleForTesting |
| 178 | Map<Integer, OkHttpClientStream> getStreams() { |
| 179 | return streams; |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 180 | } |
| 181 | |
| 182 | /** |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 183 | * Finish all active streams with given status, then close the transport. |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 184 | */ |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 185 | void abort(Status status) { |
| 186 | onGoAway(-1, status); |
| 187 | } |
| 188 | |
| 189 | private void onGoAway(int lastKnownStreamId, Status status) { |
| 190 | ArrayList<OkHttpClientStream> goAwayStreams = new ArrayList<OkHttpClientStream>(); |
| 191 | synchronized (lock) { |
| 192 | goAway = true; |
| 193 | goAwayStatus = status; |
| 194 | Iterator<Map.Entry<Integer, OkHttpClientStream>> it = streams.entrySet().iterator(); |
| 195 | while (it.hasNext()) { |
| 196 | Map.Entry<Integer, OkHttpClientStream> entry = it.next(); |
| 197 | if (entry.getKey() > lastKnownStreamId) { |
| 198 | goAwayStreams.add(entry.getValue()); |
| 199 | it.remove(); |
| 200 | } |
| 201 | } |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 202 | } |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 203 | |
| 204 | // Starting stop, go into STOPPING state so that Channel know this Transport should not be used |
| 205 | // further, will become STOPPED once all streams are complete. |
| 206 | stopAsync(); |
| 207 | |
| 208 | for (OkHttpClientStream stream : goAwayStreams) { |
lryan | e4bd1c7 | 2014-09-08 14:03:35 -0700 | [diff] [blame] | 209 | stream.setStatus(status, new Metadata.Trailers()); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 210 | } |
| 211 | } |
| 212 | |
| 213 | /** |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 214 | * Called when a stream is closed. |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 215 | * |
| 216 | * <p> Return false if the stream has already finished. |
| 217 | */ |
| 218 | private boolean finishStream(int streamId, Status status) { |
| 219 | OkHttpClientStream stream; |
| 220 | stream = streams.remove(streamId); |
| 221 | if (stream != null) { |
| 222 | // This is mainly for failed streams, for successfully finished streams, it's a no-op. |
lryan | e4bd1c7 | 2014-09-08 14:03:35 -0700 | [diff] [blame] | 223 | stream.setStatus(status, new Metadata.Trailers()); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 224 | return true; |
| 225 | } |
| 226 | return false; |
| 227 | } |
| 228 | |
| 229 | /** |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 230 | * When the transport is in goAway states, we should stop it once all active streams finish. |
| 231 | */ |
| 232 | private void stopIfNecessary() { |
| 233 | boolean shouldStop; |
| 234 | synchronized (lock) { |
| 235 | shouldStop = (goAway && streams.size() == 0); |
| 236 | } |
| 237 | if (shouldStop) { |
| 238 | frameWriter.close(); |
| 239 | try { |
| 240 | frameReader.close(); |
| 241 | } catch (IOException e) { |
| 242 | throw new RuntimeException(e); |
| 243 | } |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 244 | notifyStopped(); |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | /** |
| 249 | * Returns a Grpc status corresponding to the given ErrorCode. |
| 250 | */ |
| 251 | @VisibleForTesting |
| 252 | static Status toGrpcStatus(ErrorCode code) { |
| 253 | return ERROR_CODE_TO_STATUS.get(code); |
| 254 | } |
| 255 | |
| 256 | /** |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 257 | * Runnable which reads frames and dispatches them to in flight calls |
| 258 | */ |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 259 | @VisibleForTesting |
| 260 | class ClientFrameHandler implements FrameReader.Handler, Runnable { |
| 261 | ClientFrameHandler() {} |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 262 | |
| 263 | @Override |
| 264 | public void run() { |
| 265 | String threadName = Thread.currentThread().getName(); |
| 266 | Thread.currentThread().setName("OkHttpClientTransport"); |
| 267 | try { |
| 268 | // Read until the underlying socket closes. |
| 269 | while (frameReader.nextFrame(this)) { |
| 270 | } |
| 271 | } catch (IOException ioe) { |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 272 | abort(Status.fromThrowable(ioe)); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 273 | } finally { |
| 274 | // Restore the original thread name. |
| 275 | Thread.currentThread().setName(threadName); |
| 276 | } |
| 277 | } |
| 278 | |
| 279 | /** |
| 280 | * Handle a HTTP2 DATA frame |
| 281 | */ |
| 282 | @Override |
| 283 | public void data(boolean inFinished, int streamId, BufferedSource in, int length) |
| 284 | throws IOException { |
| 285 | final OkHttpClientStream stream; |
| 286 | stream = streams.get(streamId); |
| 287 | if (stream == null) { |
| 288 | frameWriter.rstStream(streamId, ErrorCode.INVALID_STREAM); |
| 289 | return; |
| 290 | } |
| 291 | InputStreamDeframer deframer = stream.getDeframer(); |
| 292 | |
| 293 | // Wait until the frame is complete. |
| 294 | in.require(length); |
| 295 | |
| 296 | deframer.deliverFrame(ByteStreams.limit(in.inputStream(), length), inFinished); |
| 297 | unacknowledgedBytesRead += length; |
| 298 | stream.unacknowledgedBytesRead += length; |
| 299 | if (unacknowledgedBytesRead >= DEFAULT_INITIAL_WINDOW_SIZE / 2) { |
| 300 | frameWriter.windowUpdate(0, unacknowledgedBytesRead); |
| 301 | unacknowledgedBytesRead = 0; |
| 302 | } |
| 303 | if (stream.unacknowledgedBytesRead >= DEFAULT_INITIAL_WINDOW_SIZE / 2) { |
| 304 | frameWriter.windowUpdate(streamId, stream.unacknowledgedBytesRead); |
| 305 | stream.unacknowledgedBytesRead = 0; |
| 306 | } |
| 307 | if (inFinished) { |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 308 | if (finishStream(streamId, Status.OK)) { |
| 309 | stopIfNecessary(); |
| 310 | } |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 311 | } |
| 312 | } |
| 313 | |
| 314 | /** |
| 315 | * Handle HTTP2 HEADER and CONTINUATION frames |
| 316 | */ |
| 317 | @Override |
| 318 | public void headers(boolean outFinished, |
| 319 | boolean inFinished, |
| 320 | int streamId, |
| 321 | int associatedStreamId, |
brettmorgan | 56f5ec3 | 2014-07-18 16:54:50 -0700 | [diff] [blame] | 322 | List<Header> headerBlock, |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 323 | HeadersMode headersMode) { |
| 324 | // TODO(user): handle received headers. |
| 325 | } |
| 326 | |
| 327 | @Override |
| 328 | public void rstStream(int streamId, ErrorCode errorCode) { |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 329 | if (finishStream(streamId, toGrpcStatus(errorCode))) { |
| 330 | stopIfNecessary(); |
| 331 | } |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 332 | } |
| 333 | |
| 334 | @Override |
| 335 | public void settings(boolean clearPrevious, Settings settings) { |
| 336 | // not impl |
| 337 | frameWriter.ackSettings(); |
| 338 | } |
| 339 | |
| 340 | @Override |
| 341 | public void ping(boolean ack, int payload1, int payload2) { |
| 342 | if (!ack) { |
| 343 | frameWriter.ping(true, payload1, payload2); |
| 344 | } |
| 345 | } |
| 346 | |
| 347 | @Override |
| 348 | public void ackSettings() { |
| 349 | // Do nothing currently. |
| 350 | } |
| 351 | |
| 352 | @Override |
| 353 | public void goAway(int lastGoodStreamId, ErrorCode errorCode, ByteString debugData) { |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 354 | onGoAway(lastGoodStreamId, Status.UNAVAILABLE.withDescription("Go away")); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 355 | } |
| 356 | |
| 357 | @Override |
| 358 | public void pushPromise(int streamId, int promisedStreamId, List<Header> requestHeaders) |
| 359 | throws IOException { |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 360 | // We don't accept server initiated stream. |
| 361 | frameWriter.rstStream(streamId, ErrorCode.PROTOCOL_ERROR); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 362 | } |
| 363 | |
| 364 | @Override |
| 365 | public void windowUpdate(int arg0, long arg1) { |
| 366 | // TODO(user): flow control. |
| 367 | } |
| 368 | |
| 369 | @Override |
brettmorgan | 56f5ec3 | 2014-07-18 16:54:50 -0700 | [diff] [blame] | 370 | public void priority(int streamId, int streamDependency, int weight, boolean exclusive) { |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 371 | // Ignore priority change. |
| 372 | // TODO(user): log |
| 373 | } |
brettmorgan | 56f5ec3 | 2014-07-18 16:54:50 -0700 | [diff] [blame] | 374 | |
| 375 | @Override |
| 376 | public void alternateService(int streamId, String origin, ByteString protocol, String host, |
| 377 | int port, long maxAge) { |
| 378 | // TODO(user): Deal with alternateService propagation |
| 379 | } |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 380 | } |
| 381 | |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 382 | @GuardedBy("lock") |
| 383 | private void assignStreamId(OkHttpClientStream stream) { |
| 384 | Preconditions.checkState(stream.streamId == 0, "StreamId already assigned"); |
| 385 | stream.streamId = nextStreamId; |
| 386 | streams.put(stream.streamId, stream); |
| 387 | if (nextStreamId >= Integer.MAX_VALUE - 2) { |
lryan | 71e4a92 | 2014-09-25 18:25:54 -0700 | [diff] [blame^] | 388 | onGoAway(Integer.MAX_VALUE, Status.INTERNAL.withDescription("Stream id exhaust")); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 389 | } else { |
| 390 | nextStreamId += 2; |
| 391 | } |
| 392 | } |
| 393 | |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 394 | /** |
| 395 | * Client stream for the okhttp transport. |
| 396 | */ |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 397 | @VisibleForTesting |
zhangkun | 048649e | 2014-08-28 15:52:03 -0700 | [diff] [blame] | 398 | class OkHttpClientStream extends AbstractClientStream { |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 399 | int streamId; |
| 400 | final InputStreamDeframer deframer; |
| 401 | int unacknowledgedBytesRead; |
| 402 | |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 403 | OkHttpClientStream(MethodDescriptor<?, ?> method, Metadata.Headers headers, |
ejona | 6fc356b | 2014-09-22 12:49:20 -0700 | [diff] [blame] | 404 | ClientStreamListener listener) { |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 405 | super(listener); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 406 | deframer = new InputStreamDeframer(inboundMessageHandler()); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 407 | synchronized (lock) { |
| 408 | if (goAway) { |
lryan | e4bd1c7 | 2014-09-08 14:03:35 -0700 | [diff] [blame] | 409 | setStatus(goAwayStatus, new Metadata.Trailers()); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 410 | return; |
| 411 | } |
| 412 | assignStreamId(this); |
| 413 | } |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 414 | String defaultPath = "/" + method.getName(); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 415 | frameWriter.synStream(false, false, streamId, 0, |
nathanmittler | 23fbc7c | 2014-09-11 12:50:16 -0700 | [diff] [blame] | 416 | Headers.createRequestHeaders(headers, defaultPath, defaultAuthority)); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 417 | } |
| 418 | |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 419 | InputStreamDeframer getDeframer() { |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 420 | return deframer; |
| 421 | } |
| 422 | |
| 423 | @Override |
| 424 | protected void sendFrame(ByteBuffer frame, boolean endOfStream) { |
| 425 | Preconditions.checkState(streamId != 0, "streamId should be set"); |
simonma | 18eb63b | 2014-09-04 13:16:28 -0700 | [diff] [blame] | 426 | Buffer buffer = new Buffer(); |
| 427 | // Read the data into a buffer. |
| 428 | // TODO(user): swap to NIO buffers or zero-copy if/when okhttp/okio supports it |
| 429 | buffer.write(frame.array(), frame.arrayOffset(), frame.remaining()); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 430 | // Write the data to the remote endpoint. |
| 431 | frameWriter.data(endOfStream, streamId, buffer); |
| 432 | frameWriter.flush(); |
| 433 | } |
| 434 | |
| 435 | @Override |
nathanmittler | 43d2fcc | 2014-08-29 21:58:54 -0700 | [diff] [blame] | 436 | protected void disableWindowUpdate(ListenableFuture<Void> processingFuture) { |
| 437 | // TODO(user): implement inbound flow control. |
| 438 | } |
| 439 | |
| 440 | @Override |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 441 | public void cancel() { |
simonma | 1966d5b | 2014-08-08 14:51:01 -0700 | [diff] [blame] | 442 | if (streamId == 0) { |
| 443 | // This should only happens when the stream was failed in constructor. |
| 444 | Preconditions.checkState(state() == StreamState.CLOSED, "A unclosed stream has no id"); |
| 445 | } |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 446 | outboundPhase = Phase.STATUS; |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 447 | if (finishStream(streamId, toGrpcStatus(ErrorCode.CANCEL))) { |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 448 | frameWriter.rstStream(streamId, ErrorCode.CANCEL); |
simonma | 7bf17dc | 2014-07-29 09:52:20 -0700 | [diff] [blame] | 449 | stopIfNecessary(); |
simonma | 77d9706 | 2014-07-18 10:22:35 -0700 | [diff] [blame] | 450 | } |
| 451 | } |
| 452 | } |
| 453 | } |