You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ol...@apache.org on 2016/09/18 11:29:33 UTC
svn commit: r1761318 [2/8] - in /httpcomponents/httpcore/trunk:
httpcore5-h2/src/main/java/org/apache/hc/core5/http2/
httpcore5-h2/src/main/java/org/apache/hc/core5/http2/config/
httpcore5-h2/src/main/java/org/apache/hc/core5/http2/frame/ httpcore5-h2/...
Added: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractHttp2StreamMultiplexer.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractHttp2StreamMultiplexer.java?rev=1761318&view=auto
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractHttp2StreamMultiplexer.java (added)
+++ httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractHttp2StreamMultiplexer.java Sun Sep 18 11:29:30 2016
@@ -0,0 +1,1355 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.http2.impl.nio;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.hc.core5.http.ConnectionClosedException;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.ProtocolException;
+import org.apache.hc.core5.http2.H2ConnectionException;
+import org.apache.hc.core5.http2.H2Error;
+import org.apache.hc.core5.http2.H2StreamResetException;
+import org.apache.hc.core5.http2.config.H2Config;
+import org.apache.hc.core5.http2.config.H2Param;
+import org.apache.hc.core5.http2.config.H2Setting;
+import org.apache.hc.core5.http2.frame.FrameFactory;
+import org.apache.hc.core5.http2.frame.FrameFlag;
+import org.apache.hc.core5.http2.frame.FrameType;
+import org.apache.hc.core5.http2.frame.RawFrame;
+import org.apache.hc.core5.http2.frame.StreamIdGenerator;
+import org.apache.hc.core5.http2.hpack.HPackDecoder;
+import org.apache.hc.core5.http2.hpack.HPackEncoder;
+import org.apache.hc.core5.http2.impl.BasicH2TransportMetrics;
+import org.apache.hc.core5.http2.nio.AsyncPushProducer;
+import org.apache.hc.core5.http2.nio.command.ExecutionCommand;
+import org.apache.hc.core5.http2.nio.command.ShutdownCommand;
+import org.apache.hc.core5.http2.nio.command.ShutdownType;
+import org.apache.hc.core5.reactor.Command;
+import org.apache.hc.core5.reactor.IOSession;
+import org.apache.hc.core5.util.Args;
+import org.apache.hc.core5.util.ByteArrayBuffer;
+import org.apache.hc.core5.util.NetUtils;
+
+abstract class AbstractHttp2StreamMultiplexer {
+
+ private static final long LINGER_TIME = 1000; // 1 second
+
+ enum Mode { CLIENT, SERVER}
+ enum ConnectionHandshake { READY, ACTIVE, GRACEFUL_SHUTDOWN, SHUTDOWN}
+ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
+
+ private final Mode mode;
+ private final IOSession ioSession;
+ private final FrameFactory frameFactory;
+ private final StreamIdGenerator idGenerator;
+ private final H2Config localConfig;
+ private final BasicH2TransportMetrics metrics;
+ private final FrameInputBuffer inputBuffer;
+ private final FrameOutputBuffer outputBuffer;
+ private final Deque<RawFrame> outputQueue;
+ private final HPackEncoder hPackEncoder;
+ private final HPackDecoder hPackDecoder;
+ private final Map<Integer, Http2Stream> streamMap;
+ private final AtomicInteger connInputWindow;
+ private final AtomicInteger connOutputWindow;
+ private final Lock outputLock;
+ private final AtomicInteger outputRequests;
+ private final AtomicInteger lastStreamId;
+ private final Http2StreamListener callback;
+
+ private ConnectionHandshake connState = ConnectionHandshake.READY;
+ private SettingsHandshake localSettingState = SettingsHandshake.READY;
+ private SettingsHandshake remoteSettingState = SettingsHandshake.READY;
+ private H2Config remoteConfig;
+ private int lowMark;
+
+ private Continuation continuation;
+
+ private int processedRemoteStreamId;
+
+ AbstractHttp2StreamMultiplexer(
+ final Mode mode,
+ final IOSession ioSession,
+ final FrameFactory frameFactory,
+ final StreamIdGenerator idGenerator,
+ final Charset charset,
+ final H2Config h2Config,
+ final Http2StreamListener callback) {
+ this.mode = Args.notNull(mode, "Mode");
+ this.ioSession = Args.notNull(ioSession, "IO session");
+ this.frameFactory = Args.notNull(frameFactory, "Frame factory");
+ this.idGenerator = Args.notNull(idGenerator, "Stream id generator");
+ this.localConfig = h2Config != null ? h2Config : H2Config.DEFAULT;
+ this.metrics = new BasicH2TransportMetrics();
+ this.inputBuffer = new FrameInputBuffer(this.metrics, this.localConfig.getMaxFrameSize());
+ this.outputBuffer = new FrameOutputBuffer(this.metrics, this.localConfig.getMaxFrameSize());
+ this.outputQueue = new ConcurrentLinkedDeque<>();
+ this.outputLock = new ReentrantLock();
+ this.outputRequests = new AtomicInteger(0);
+ this.lastStreamId = new AtomicInteger(0);
+ this.hPackEncoder = new HPackEncoder(charset != null ? charset : StandardCharsets.US_ASCII);
+ this.hPackDecoder = new HPackDecoder(charset != null ? charset : StandardCharsets.US_ASCII);
+ this.streamMap = new ConcurrentHashMap<>();
+ this.connInputWindow = new AtomicInteger(localConfig.getInitialWindowSize());
+ this.connOutputWindow = new AtomicInteger(H2Config.DEFAULT.getInitialWindowSize());
+
+ this.hPackDecoder.setMaxTableSize(H2Config.DEFAULT.getInitialWindowSize());
+ this.hPackEncoder.setMaxTableSize(this.localConfig.getHeaderTableSize());
+
+ this.remoteConfig = H2Config.DEFAULT;
+ this.lowMark = this.remoteConfig.getInitialWindowSize() / 2;
+ this.callback = callback;
+ }
+
+ abstract Http2StreamHandler createRemotelyInitiatedStream(Http2StreamChannel channel) throws IOException;
+
+ private int updateWindow(final AtomicInteger window, final int delta) throws ArithmeticException {
+ for (;;) {
+ final int current = window.get();
+ final long newValue = current + delta;
+ if (Math.abs(newValue) > Integer.MAX_VALUE) {
+ throw new ArithmeticException("Update causes flow control window to exceed " + Integer.MAX_VALUE);
+ }
+ if (window.compareAndSet(current, (int) newValue)) {
+ return (int) newValue;
+ }
+ }
+ }
+
+ private int updateInputWindow(
+ final int streamId, final AtomicInteger window, final int delta) throws ArithmeticException {
+ final int newSize = updateWindow(window, delta);
+ if (callback != null) {
+ callback.onInputFlowControl(streamId, delta, newSize);
+ }
+ return newSize;
+ }
+
+ private int updateOutputWindow(
+ final int streamId, final AtomicInteger window, final int delta) throws ArithmeticException {
+ final int newSize = updateWindow(window, delta);
+ if (callback != null) {
+ callback.onOutputFlowControl(streamId, delta, newSize);
+ }
+ return newSize;
+ }
+
+ private void commitFrameInternal(final RawFrame frame) throws IOException {
+ if (outputBuffer.isEmpty() && outputQueue.isEmpty()) {
+ if (callback != null) {
+ callback.onFrameOutput(frame);
+ }
+ outputBuffer.write(frame, ioSession.channel());
+ } else {
+ outputQueue.addLast(frame);
+ }
+ ioSession.setEvent(SelectionKey.OP_WRITE);
+ }
+
+ private void commitFrame(final RawFrame frame) throws IOException {
+ Args.notNull(frame, "Frame");
+ outputLock.lock();
+ try {
+ commitFrameInternal(frame);
+ } finally {
+ outputLock.unlock();
+ }
+ }
+
+ private void commitHeaders(
+ final int streamId, final List<Header> headers, final boolean endStream) throws IOException {
+ if (headers == null || headers.isEmpty()) {
+ throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Message headers are missing");
+ }
+ if (callback != null) {
+ callback.onHeaderOutput(headers);
+ }
+ final ByteArrayBuffer buf = new ByteArrayBuffer(512);
+ hPackEncoder.encodeHeaders(buf, headers);
+
+ int off = 0;
+ int remaining = buf.length();
+ boolean continuation = false;
+
+ while (remaining > 0) {
+ final int chunk = Math.min(remoteConfig.getMaxFrameSize(), remaining);
+ final ByteBuffer payload = ByteBuffer.wrap(buf.array(), off, chunk);
+
+ remaining -= chunk;
+ off += chunk;
+
+ final boolean endHeaders = remaining == 0;
+ final RawFrame frame;
+ if (!continuation) {
+ frame = frameFactory.createHeaders(streamId, payload, endHeaders, endStream);
+ continuation = true;
+ } else {
+ frame = frameFactory.createContinuation(streamId, payload, endHeaders);
+ }
+ commitFrameInternal(frame);
+ }
+ }
+
+ private void commitPushPromise(
+ final int streamId, final int promisedStreamId, final List<Header> headers) throws IOException {
+ if (headers == null || headers.isEmpty()) {
+ throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Message headers are missing");
+ }
+ if (callback != null) {
+ callback.onHeaderOutput(headers);
+ }
+ final ByteArrayBuffer buf = new ByteArrayBuffer(512);
+ buf.append((byte)(promisedStreamId >> 24));
+ buf.append((byte)(promisedStreamId >> 16));
+ buf.append((byte)(promisedStreamId >> 8));
+ buf.append((byte)(promisedStreamId));
+
+ hPackEncoder.encodeHeaders(buf, headers);
+
+ int off = 0;
+ int remaining = buf.length();
+ boolean continuation = false;
+
+ while (remaining > 0) {
+ final int chunk = Math.min(remoteConfig.getMaxFrameSize(), remaining);
+ final ByteBuffer payload = ByteBuffer.wrap(buf.array(), off, chunk);
+
+ remaining -= chunk;
+ off += chunk;
+
+ final boolean endHeaders = remaining == 0;
+ final RawFrame frame;
+ if (!continuation) {
+ frame = frameFactory.createPushPromise(streamId, payload, endHeaders);
+ continuation = true;
+ } else {
+ frame = frameFactory.createContinuation(streamId, payload, endHeaders);
+ }
+ commitFrameInternal(frame);
+ }
+ }
+
+ private int streamData(
+ final int streamId, final AtomicInteger streamOutputWindow, final ByteBuffer payload) throws IOException {
+ if (outputBuffer.isEmpty() && outputQueue.isEmpty()) {
+ final int capacity = Math.min(connOutputWindow.get(), streamOutputWindow.get());
+ if (capacity <= 0) {
+ return 0;
+ }
+ final int frameSize = Math.max(localConfig.getMaxFrameSize(), remoteConfig.getMaxFrameSize());
+ final int maxPayloadSize = Math.min(capacity, frameSize);
+ final int chunk;
+ if (payload.remaining() <= maxPayloadSize) {
+ chunk = payload.remaining();
+ final RawFrame dataFrame = frameFactory.createData(streamId, payload, false);
+ if (callback != null) {
+ callback.onFrameOutput(dataFrame);
+ }
+ outputBuffer.write(dataFrame, ioSession.channel());
+ } else {
+ chunk = maxPayloadSize;
+ final int originalLimit = payload.limit();
+ try {
+ payload.limit(payload.position() + chunk);
+ final RawFrame dataFrame = frameFactory.createData(streamId, payload, false);
+ if (callback != null) {
+ callback.onFrameOutput(dataFrame);
+ }
+ outputBuffer.write(dataFrame, ioSession.channel());
+ } finally {
+ payload.limit(originalLimit);
+ }
+ }
+ payload.position(payload.position() + chunk);
+ updateOutputWindow(0, connOutputWindow, -chunk);
+ updateOutputWindow(streamId, streamOutputWindow, -chunk);
+ ioSession.setEvent(SelectionKey.OP_WRITE);
+ return chunk;
+ } else {
+ return 0;
+ }
+ }
+
+ private void updateInputCapacity(
+ final int streamId, final AtomicInteger inputWindow, final int inputCapacity) throws IOException {
+ if (inputCapacity > 0) {
+ final int streamWinSize = inputWindow.get();
+ final int chunk = inputCapacity - streamWinSize;
+ if (chunk > 0) {
+ final RawFrame windowUpdateFrame = frameFactory.createWindowUpdate(streamId, chunk);
+ commitFrame(windowUpdateFrame);
+ updateInputWindow(streamId, inputWindow, chunk);
+ }
+ }
+ }
+
+ private void requestSessionOutput() {
+ outputRequests.incrementAndGet();
+ ioSession.setEvent(SelectionKey.OP_WRITE);
+ }
+
+ private void updateLastStreamId(final int streamId) {
+ final int currentId = lastStreamId.get();
+ if (streamId > currentId) {
+ lastStreamId.compareAndSet(currentId, streamId);
+ }
+ }
+
+ private int generateStreamId() {
+ for (;;) {
+ final int currentId = lastStreamId.get();
+ final int newStreamId = idGenerator.generate(currentId);
+ if (lastStreamId.compareAndSet(currentId, newStreamId)) {
+ return newStreamId;
+ }
+ }
+ }
+
+ public final void onConnect(final ByteBuffer prefeed) throws HttpException, IOException {
+ if (prefeed != null) {
+ inputBuffer.put(prefeed);
+ }
+ connState = ConnectionHandshake.ACTIVE;
+ final RawFrame settingsFrame = frameFactory.createSettings(
+ new H2Setting(H2Param.HEADER_TABLE_SIZE, localConfig.getHeaderTableSize()),
+ new H2Setting(H2Param.ENABLE_PUSH, localConfig.isPushEnabled() ? 1 : 0),
+ new H2Setting(H2Param.MAX_CONCURRENT_STREAMS, localConfig.getMaxConcurrentStreams()),
+ new H2Setting(H2Param.INITIAL_WINDOW_SIZE, localConfig.getInitialWindowSize()),
+ new H2Setting(H2Param.MAX_FRAME_SIZE, localConfig.getMaxFrameSize()),
+ new H2Setting(H2Param.MAX_HEADER_LIST_SIZE, localConfig.getMaxHeaderListSize()));
+
+ commitFrame(settingsFrame);
+ localSettingState = SettingsHandshake.TRANSMITTED;
+ }
+
+ public final void onInput() throws HttpException, IOException {
+ if (connState == ConnectionHandshake.SHUTDOWN) {
+ ioSession.clearEvent(SelectionKey.OP_READ);
+ } else {
+ RawFrame frame;
+ while ((frame = inputBuffer.read(ioSession.channel())) != null) {
+ if (callback != null) {
+ callback.onFrameInput(frame);
+ }
+ consumeFrame(frame);
+ }
+ }
+ }
+
+ public final void onOutput() throws HttpException, IOException {
+ outputLock.lock();
+ try {
+ if (!outputBuffer.isEmpty()) {
+ outputBuffer.flush(ioSession.channel());
+ }
+ while (outputBuffer.isEmpty()) {
+ final RawFrame frame = outputQueue.poll();
+ if (frame != null) {
+ if (callback != null) {
+ callback.onFrameOutput(frame);
+ }
+ outputBuffer.write(frame, ioSession.channel());
+ } else {
+ break;
+ }
+ }
+ } finally {
+ outputLock.unlock();
+ }
+
+ switch (connState) {
+ case ACTIVE:
+ processPendingCommands();
+ break;
+ case GRACEFUL_SHUTDOWN:
+ case SHUTDOWN:
+ cancelPendingCommands();
+ break;
+ }
+
+ final int connWinSize = connInputWindow.get();
+ if (connWinSize < lowMark) {
+ final int delta = this.remoteConfig.getInitialWindowSize() - connWinSize;
+ if (delta > 0) {
+ final RawFrame windowUpdateFrame = frameFactory.createWindowUpdate(0, delta);
+ commitFrame(windowUpdateFrame);
+ updateInputWindow(0, connInputWindow, delta);
+ }
+ }
+
+ if (connState != ConnectionHandshake.SHUTDOWN && remoteSettingState == SettingsHandshake.ACKED) {
+
+ if (connOutputWindow.get() > 0) {
+ produceOutput();
+ }
+ final int pendingOutputRequests = outputRequests.get();
+ boolean outputPending = false;
+ if (!streamMap.isEmpty() && connOutputWindow.get() > 0) {
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final Http2Stream stream = entry.getValue();
+ if (!stream.isLocalClosed()
+ && stream.getOutputWindow().get() > 0
+ && stream.isOutputReady()) {
+ outputPending = true;
+ break;
+ }
+ }
+ }
+ if (!outputPending) {
+ outputLock.lock();
+ try {
+ if (!outputBuffer.isEmpty() || !outputQueue.isEmpty()) {
+ outputPending = true;
+ }
+ } finally {
+ outputLock.unlock();
+ }
+ }
+ if (!outputPending && outputRequests.compareAndSet(pendingOutputRequests, 0)) {
+ ioSession.clearEvent(SelectionKey.OP_WRITE);
+ } else {
+ outputRequests.addAndGet(-pendingOutputRequests);
+ }
+ }
+
+ if (connState == ConnectionHandshake.SHUTDOWN) {
+ outputLock.lock();
+ try {
+ if (outputBuffer.isEmpty() && outputQueue.isEmpty()) {
+ ioSession.close();
+ }
+ } finally {
+ outputLock.unlock();
+ }
+ }
+ }
+
+ public final void onTimeout() throws HttpException, IOException {
+ connState = ConnectionHandshake.SHUTDOWN;
+
+ final RawFrame goAway;
+ if (localSettingState != SettingsHandshake.ACKED) {
+ goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.SETTINGS_TIMEOUT, "Setting timeout");
+ } else {
+ goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR, "Timeout due to inactivity");
+ }
+ commitFrame(goAway);
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final Http2Stream stream = entry.getValue();
+ stream.reset(new H2StreamResetException(H2Error.NO_ERROR, "Timeout due to inactivity"));
+ }
+ streamMap.clear();
+ }
+
+ private void processPendingCommands() throws IOException, HttpException {
+ final Queue<Command> commandQueue = ioSession.getCommandQueue();
+ while (!commandQueue.isEmpty() && streamMap.size() < remoteConfig.getMaxConcurrentStreams()) {
+ final Command command = commandQueue.remove();
+ if (command instanceof ShutdownCommand) {
+ final ShutdownCommand shutdownCommand = (ShutdownCommand) command;
+ if (shutdownCommand.getType() == ShutdownType.IMMEDIATE) {
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final Http2Stream stream = entry.getValue();
+ stream.cancel();
+ }
+ streamMap.clear();
+ connState = ConnectionHandshake.SHUTDOWN;
+ } else {
+ if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
+ connState = ConnectionHandshake.GRACEFUL_SHUTDOWN;
+ final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR, "Graceful shutdown");
+ commitFrame(goAway);
+ }
+ }
+ break;
+ } else if (command instanceof ExecutionCommand) {
+ if (mode == Mode.SERVER) {
+ throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Illegal attempt to execute a request");
+ }
+ @SuppressWarnings("unchecked")
+ final ExecutionCommand<Object> executionCommand = (ExecutionCommand<Object>) command;
+ final int streamId = generateStreamId();
+ final Http2StreamChannelImpl channel = new Http2StreamChannelImpl(
+ streamId,
+ localConfig.getInitialWindowSize(),
+ remoteConfig.getInitialWindowSize());
+ final Http2StreamHandler streamHandler = new ClientHttp2StreamHandler<>(
+ channel,
+ executionCommand.getRequestProducer(),
+ executionCommand.getResponseConsumer(),
+ executionCommand.getCallback());
+ final Http2Stream stream = new Http2Stream(channel, streamHandler, false);
+ if (stream.isOutputReady()) {
+ stream.produceOutput();
+ }
+
+ streamMap.put(streamId, stream);
+ if (!outputQueue.isEmpty()) {
+ return;
+ }
+ }
+ }
+ }
+
+ private void cancelPendingCommands() {
+ final Queue<Command> commandQueue = ioSession.getCommandQueue();
+ while (!commandQueue.isEmpty()) {
+ final Command command = commandQueue.remove();
+ command.cancel();
+ }
+ }
+
+ public final void onException(final Exception cause) {
+ try {
+ if (!(cause instanceof ConnectionClosedException)) {
+ final H2Error errorCode;
+ if (cause instanceof H2ConnectionException) {
+ errorCode = H2Error.getByCode(((H2ConnectionException) cause).getCode());
+ } else if (cause instanceof ProtocolException){
+ errorCode = H2Error.PROTOCOL_ERROR;
+ } else {
+ errorCode = H2Error.INTERNAL_ERROR;
+ }
+ final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, errorCode, cause.getMessage());
+ commitFrame(goAway);
+ }
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final Http2Stream stream = entry.getValue();
+ stream.reset(cause);
+ }
+ streamMap.clear();
+ cancelPendingCommands();
+ connState = ConnectionHandshake.SHUTDOWN;
+ } catch (IOException ignore) {
+ } finally {
+ ioSession.shutdown();
+ }
+ }
+
+ private Http2Stream getValidStream(final int streamId) throws H2ConnectionException {
+ if (streamId == 0) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id: " + streamId);
+ }
+ final Http2Stream stream = streamMap.get(streamId);
+ if (stream == null) {
+ if (streamId <= lastStreamId.get()) {
+ throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
+ } else {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected stream id: " + streamId);
+ }
+ }
+ return stream;
+ }
+
+ private void consumeFrame(final RawFrame frame) throws HttpException, IOException {
+ final FrameType frameType = FrameType.valueOf(frame.getType());
+ final int streamId = frame.getStreamId();
+ if (continuation != null && frameType != FrameType.CONTINUATION) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "CONTINUATION frame expected");
+ }
+ if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) >= 0) {
+ if (streamId > processedRemoteStreamId && !idGenerator.isSameSide(streamId)) {
+ // ignore the frame
+ return;
+ }
+ }
+ switch (frameType) {
+ case DATA: {
+ final Http2Stream stream = getValidStream(streamId);
+ try {
+ consumeDataFrame(frame, stream);
+ } catch (H2StreamResetException ex) {
+ stream.localReset(ex);
+ }
+
+ if (stream.isTerminated()) {
+ streamMap.remove(streamId);
+ stream.releaseResources();
+ }
+ }
+ break;
+ case HEADERS: {
+ if (streamId == 0) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id: " + streamId);
+ }
+ Http2Stream stream = streamMap.get(streamId);
+ if (stream == null) {
+ if (mode != Mode.SERVER) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal HEADERS frame");
+ }
+
+ updateLastStreamId(streamId);
+
+ final Http2StreamChannelImpl channel = new Http2StreamChannelImpl(
+ streamId,
+ localConfig.getInitialWindowSize(),
+ remoteConfig.getInitialWindowSize());
+ final Http2StreamHandler streamHandler = createRemotelyInitiatedStream(channel);
+ stream = new Http2Stream(channel, streamHandler, true);
+ if (stream.isOutputReady()) {
+ stream.produceOutput();
+ }
+ streamMap.put(streamId, stream);
+ }
+
+ try {
+ consumeHeaderFrame(frame, stream);
+
+ if (stream.isOutputReady()) {
+ stream.produceOutput();
+ }
+ } catch (H2StreamResetException ex) {
+ stream.localReset(ex);
+ }
+
+ if (stream.isTerminated()) {
+ streamMap.remove(streamId);
+ stream.releaseResources();
+ }
+ }
+ break;
+ case CONTINUATION: {
+ if (continuation == null) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected CONTINUATION frame");
+ }
+ if (streamId != continuation.streamId) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected CONTINUATION stream id: " + streamId);
+ }
+
+ final Http2Stream stream = getValidStream(streamId);
+ try {
+
+ consumeContinuationFrame(frame, stream);
+ } catch (H2StreamResetException ex) {
+ stream.localReset(ex);
+ }
+
+ if (stream.isTerminated()) {
+ streamMap.remove(streamId);
+ stream.releaseResources();
+ }
+ }
+ break;
+ case WINDOW_UPDATE: {
+ final ByteBuffer payload = frame.getPayload();
+ if (payload == null || payload.remaining() != 4) {
+ throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid WINDOW_UPDATE frame payload");
+ }
+ final int delta = payload.getInt();
+ if (delta <= 0) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Invalid WINDOW_UPDATE delta");
+ }
+ if (streamId == 0) {
+ try {
+ updateOutputWindow(0, connOutputWindow, delta);
+ } catch (ArithmeticException ex) {
+ throw new H2ConnectionException(H2Error.FLOW_CONTROL_ERROR, ex.getMessage());
+ }
+ } else {
+ final Http2Stream stream = streamMap.get(streamId);
+ if (stream != null) {
+ try {
+ updateOutputWindow(streamId, stream.getOutputWindow(), delta);
+ } catch (ArithmeticException ex) {
+ throw new H2ConnectionException(H2Error.FLOW_CONTROL_ERROR, ex.getMessage());
+ }
+ }
+ }
+ ioSession.setEvent(SelectionKey.OP_WRITE);
+ }
+ break;
+ case RST_STREAM: {
+ if (streamId == 0) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id: " + streamId);
+ }
+ final Http2Stream stream = getValidStream(streamId);
+ final ByteBuffer payload = frame.getPayload();
+ if (payload == null || payload.remaining() != 4) {
+ throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid RST_STREAM frame payload");
+ }
+
+ final int errorCode = payload.getInt();
+ stream.reset(new H2StreamResetException(errorCode, "Stream reset"));
+ streamMap.remove(streamId);
+ stream.releaseResources();
+ }
+ break;
+ case PING: {
+ if (streamId != 0) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
+ }
+ if (!frame.isFlagSet(FrameFlag.ACK)) {
+ final ByteBuffer payload = frame.getPayload();
+ final RawFrame response = frameFactory.createPingAck(payload);
+ commitFrame(response);
+ }
+ }
+ break;
+ case SETTINGS: {
+ if (streamId != 0) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
+ }
+ if (frame.isFlagSet(FrameFlag.ACK)) {
+ if (localSettingState == SettingsHandshake.TRANSMITTED) {
+ localSettingState = SettingsHandshake.ACKED;
+ ioSession.setEvent(SelectionKey.OP_WRITE);
+ }
+ } else {
+ final ByteBuffer payload = frame.getPayload();
+ if (payload != null) {
+ if ((payload.remaining() % 6) != 0) {
+ throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid SETTINGS payload");
+ }
+ consumeSettingsFrame(payload);
+ remoteSettingState = SettingsHandshake.TRANSMITTED;
+ }
+ // Send ACK
+ final RawFrame response = frameFactory.createSettingsAck();
+ commitFrame(response);
+ remoteSettingState = SettingsHandshake.ACKED;
+ }
+ }
+ break;
+ case PRIORITY:
+ // Stream priority not supported
+ break;
+ case PUSH_PROMISE: {
+ if (mode == Mode.SERVER) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Push not supported");
+ }
+ if (!localConfig.isPushEnabled()) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Push is disabled");
+ }
+
+ final Http2Stream stream = getValidStream(streamId);
+ if (stream.isRemoteClosed()) {
+ stream.localReset(new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream closed"));
+ break;
+ }
+
+ final ByteBuffer payload = frame.getPayloadContent();
+ if (payload == null || payload.remaining() < 4) {
+ throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid PUSH_PROMISE payload");
+ }
+ final int promisedStreamId = payload.getInt();
+ if (promisedStreamId == 0 || idGenerator.isSameSide(promisedStreamId)) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal promised stream id: " + promisedStreamId);
+ }
+ if (streamMap.get(promisedStreamId) != null) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Unexpected promised stream id: " + promisedStreamId);
+ }
+
+ updateLastStreamId(promisedStreamId);
+
+ final Http2StreamChannelImpl channel = new Http2StreamChannelImpl(
+ promisedStreamId,
+ localConfig.getInitialWindowSize(),
+ remoteConfig.getInitialWindowSize());
+ final Http2StreamHandler streamHandler = createRemotelyInitiatedStream(channel);
+ final Http2Stream promisedStream = new Http2Stream(channel, streamHandler, true);
+ streamMap.put(promisedStreamId, promisedStream);
+
+ try {
+ consumePushPromiseFrame(frame, payload, promisedStream);
+ } catch (H2StreamResetException ex) {
+ promisedStream.localReset(ex);
+ }
+ }
+ break;
+ case GOAWAY: {
+ if (streamId != 0) {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
+ }
+ final ByteBuffer payload = frame.getPayload();
+ if (payload == null || payload.remaining() < 8) {
+ throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid RST_STREAM payload");
+ }
+ final int processedLocalStreamId = payload.getInt();
+ final int errorCode = payload.getInt();
+ if (errorCode == H2Error.NO_ERROR.getCode()) {
+ if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
+ connState = ConnectionHandshake.GRACEFUL_SHUTDOWN;
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final int activeStreamId = entry.getKey();
+ if (!idGenerator.isSameSide(activeStreamId) && activeStreamId > processedLocalStreamId) {
+ final Http2Stream stream = entry.getValue();
+ stream.cancel();
+ it.remove();
+ }
+ }
+ }
+ } else {
+ connState = ConnectionHandshake.SHUTDOWN;
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final Http2Stream stream = entry.getValue();
+ stream.reset(new H2StreamResetException(errorCode, "Connection terminated by the peer"));
+ }
+ streamMap.clear();
+ }
+ }
+ break;
+ }
+ }
+
+ private void consumeDataFrame(final RawFrame frame, final Http2Stream stream) throws HttpException, IOException {
+ final int streamId = stream.getId();
+ final ByteBuffer payload = frame.getPayloadContent();
+ if (payload != null) {
+ final int frameLength = frame.getLength();
+ final int streamWinSize = updateInputWindow(streamId, stream.getInputWindow(), -frameLength);
+ if (streamWinSize < lowMark && !stream.isRemoteClosed()) {
+ stream.produceInputCapacityUpdate();
+ }
+ final int connWinSize = updateInputWindow(0, connInputWindow, -frameLength);
+ if (connWinSize < lowMark) {
+ final int chunk = Integer.MAX_VALUE - connWinSize;
+ if (chunk > 0) {
+ final RawFrame windowUpdateFrame = frameFactory.createWindowUpdate(0, chunk);
+ commitFrame(windowUpdateFrame);
+ updateInputWindow(0, connInputWindow, chunk);
+ }
+ }
+ }
+ if (stream.isResetLocally()) {
+ return;
+ }
+ if (stream.isRemoteClosed()) {
+ throw new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream already closed");
+ }
+ if (frame.isFlagSet(FrameFlag.END_STREAM)) {
+ stream.setRemoteEndStream();
+ }
+ stream.consumeData(payload);
+ }
+
+ private void consumePushPromiseFrame(final RawFrame frame, final ByteBuffer payload, final Http2Stream promisedStream) throws HttpException, IOException {
+ final int promisedStreamId = promisedStream.getId();
+ if (!frame.isFlagSet(FrameFlag.END_HEADERS)) {
+ continuation = new Continuation(promisedStreamId, frame.getType(), true);
+ }
+ if (continuation == null) {
+ final List<Header> headers = hPackDecoder.decodeHeaders(payload);
+ if (promisedStreamId > processedRemoteStreamId) {
+ processedRemoteStreamId = promisedStreamId;
+ }
+ if (callback != null) {
+ callback.onHeaderInput(headers);
+ }
+ if (connState == ConnectionHandshake.GRACEFUL_SHUTDOWN) {
+ throw new H2StreamResetException(H2Error.REFUSED_STREAM, "Stream refused");
+ }
+ promisedStream.consumePromise(headers);
+ } else {
+ continuation.copyPayload(payload);
+ }
+ }
+
+ private void consumeHeaderFrame(final RawFrame frame, final Http2Stream stream) throws HttpException, IOException {
+ final int streamId = stream.getId();
+ if (!frame.isFlagSet(FrameFlag.END_HEADERS)) {
+ continuation = new Continuation(streamId, frame.getType(), frame.isFlagSet(FrameFlag.END_STREAM));
+ }
+ final ByteBuffer payload = frame.getPayloadContent();
+ if (frame.isFlagSet(FrameFlag.PRIORITY)) {
+ // Priority not supported
+ payload.getInt();
+ payload.get();
+ }
+ if (continuation == null) {
+ final List<Header> headers = hPackDecoder.decodeHeaders(payload);
+ if (stream.isRemoteInitiated() && streamId > processedRemoteStreamId) {
+ processedRemoteStreamId = streamId;
+ }
+ if (callback != null) {
+ callback.onHeaderInput(headers);
+ }
+ if (connState == ConnectionHandshake.GRACEFUL_SHUTDOWN) {
+ throw new H2StreamResetException(H2Error.PROTOCOL_ERROR, "Stream refused");
+ }
+ if (stream.isResetLocally()) {
+ return;
+ }
+ if (stream.isRemoteClosed()) {
+ throw new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream already closed");
+ }
+ if (frame.isFlagSet(FrameFlag.END_STREAM)) {
+ stream.setRemoteEndStream();
+ }
+ stream.consumeHeader(headers);
+ } else {
+ continuation.copyPayload(payload);
+ }
+ }
+
+ private void consumeContinuationFrame(final RawFrame frame, final Http2Stream stream) throws HttpException, IOException {
+ final int streamId = frame.getStreamId();
+ final ByteBuffer payload = frame.getPayload();
+ continuation.copyPayload(payload);
+ if (frame.isFlagSet(FrameFlag.END_HEADERS)) {
+ final List<Header> headers = hPackDecoder.decodeHeaders(continuation.getContent());
+ if (stream.isRemoteInitiated() && streamId > processedRemoteStreamId) {
+ processedRemoteStreamId = streamId;
+ }
+ if (callback != null) {
+ callback.onHeaderInput(headers);
+ }
+ if (connState == ConnectionHandshake.GRACEFUL_SHUTDOWN) {
+ throw new H2StreamResetException(H2Error.PROTOCOL_ERROR, "Stream refused");
+ }
+ if (stream.isResetLocally()) {
+ return;
+ }
+ if (stream.isRemoteClosed()) {
+ throw new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream already closed");
+ }
+ if (continuation.endStream) {
+ stream.setRemoteEndStream();
+ }
+ if (continuation.type == FrameType.PUSH_PROMISE.getValue()) {
+ stream.consumePromise(headers);
+ } else {
+ stream.consumeHeader(headers);
+ }
+ continuation = null;
+ }
+ }
+
+ private void consumeSettingsFrame(final ByteBuffer payload) throws HttpException, IOException {
+ final H2Config.Builder configBuilder = H2Config.custom();
+ while (payload.hasRemaining()) {
+ final int code = payload.getShort();
+ final H2Param param = H2Param.valueOf(code);
+ if (param != null) {
+ final int value = payload.getInt();
+ switch (param) {
+ case HEADER_TABLE_SIZE:
+ configBuilder.setHeaderTableSize(value);
+ hPackDecoder.setMaxTableSize(value);
+ break;
+ case MAX_CONCURRENT_STREAMS:
+ configBuilder.setMaxConcurrentStreams(value);
+ break;
+ case ENABLE_PUSH:
+ configBuilder.setPushEnabled(value == 1);
+ break;
+ case INITIAL_WINDOW_SIZE:
+ configBuilder.setInitialWindowSize(value);
+ final int delta = value - remoteConfig.getInitialWindowSize();
+ if (delta != 0) {
+ updateOutputWindow(0, connOutputWindow, delta);
+ if (!streamMap.isEmpty()) {
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final Http2Stream stream = entry.getValue();
+ try {
+ updateOutputWindow(stream.getId(), stream.getOutputWindow(), delta);
+ } catch (ArithmeticException ex) {
+ throw new H2ConnectionException(H2Error.FLOW_CONTROL_ERROR, ex.getMessage());
+ }
+ }
+ }
+ }
+ break;
+ case MAX_FRAME_SIZE:
+ configBuilder.setMaxFrameSize(value);
+ break;
+ case MAX_HEADER_LIST_SIZE:
+ configBuilder.setMaxHeaderListSize(value);
+ break;
+ }
+ }
+ }
+ remoteConfig = configBuilder.build();
+ lowMark = remoteConfig.getInitialWindowSize() / 2;
+ }
+
+ private void produceOutput() throws HttpException, IOException {
+ for (final Iterator<Map.Entry<Integer, Http2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
+ final Map.Entry<Integer, Http2Stream> entry = it.next();
+ final Http2Stream stream = entry.getValue();
+ if (!stream.isLocalClosed() && stream.getOutputWindow().get() > 0) {
+ stream.produceOutput();
+ }
+ if (stream.isTerminated()) {
+ it.remove();
+ stream.releaseResources();
+ }
+ if (!outputQueue.isEmpty()) {
+ break;
+ }
+ }
+
+ if (streamMap.isEmpty() && connState == ConnectionHandshake.GRACEFUL_SHUTDOWN) {
+ connState = ConnectionHandshake.SHUTDOWN;
+ }
+ }
+
+ public void requestGracefulShutdown() {
+ try {
+ final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR, "Graceful shutdown");
+ commitFrame(goAway);
+ connState = streamMap.isEmpty() ? ConnectionHandshake.SHUTDOWN : ConnectionHandshake.GRACEFUL_SHUTDOWN;
+ } catch (IOException ex) {
+ onException(ex);
+ }
+ }
+
+ @Override
+ public String toString() {
+ final SocketAddress remoteAddress = ioSession.getRemoteAddress();
+ final SocketAddress localAddress = ioSession.getLocalAddress();
+ final StringBuilder buffer = new StringBuilder();
+ NetUtils.formatAddress(buffer, localAddress);
+ buffer.append("->");
+ NetUtils.formatAddress(buffer, remoteAddress);
+ return buffer.toString();
+ }
+
+ private static class Continuation {
+
+ final int streamId;
+ final int type;
+ final boolean endStream;
+ final ByteArrayBuffer headerBuffer;
+
+ private Continuation(final int streamId, final int type, final boolean endStream) {
+ this.streamId = streamId;
+ this.type = type;
+ this.endStream = endStream;
+ this.headerBuffer = new ByteArrayBuffer(1024);
+ }
+
+ void copyPayload(final ByteBuffer payload) {
+ if (payload == null) {
+ return;
+ }
+ headerBuffer.ensureCapacity(payload.remaining());
+ payload.get(headerBuffer.array(), headerBuffer.length(), payload.remaining());
+ }
+
+ ByteBuffer getContent() {
+ return ByteBuffer.wrap(headerBuffer.array(), 0, headerBuffer.length());
+ }
+
+ }
+
+ private class Http2StreamChannelImpl implements Http2StreamChannel {
+
+ private final int id;
+ private final AtomicInteger inputWindow;
+ private final AtomicInteger outputWindow;
+
+ private volatile boolean remoteEndStream;
+ private volatile boolean localEndStream;
+
+ private volatile long deadline;
+
+ Http2StreamChannelImpl(final int id, final int initialInputWindowSize, final int initialOutputWindowSize) {
+ this.id = id;
+ this.inputWindow = new AtomicInteger(initialInputWindowSize);
+ this.outputWindow = new AtomicInteger(initialOutputWindowSize);
+ }
+
+ int getId() {
+ return id;
+ }
+
+ AtomicInteger getOutputWindow() {
+ return outputWindow;
+ }
+
+ AtomicInteger getInputWindow() {
+ return inputWindow;
+ }
+
+ @Override
+ public void submit(final List<Header> headers, final boolean endStream) throws IOException {
+ outputLock.lock();
+ try {
+ if (localEndStream) {
+ return;
+ }
+ commitHeaders(id, headers, endStream);
+ if (endStream) {
+ localEndStream = true;
+ }
+ } finally {
+ outputLock.unlock();
+ }
+ }
+
+ @Override
+ public void push(final List<Header> headers, final AsyncPushProducer pushProducer) throws HttpException, IOException {
+ if (mode == Mode.CLIENT) {
+ throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Illegal attempt to pushPromise a response");
+ }
+ final int promisedStreamId = generateStreamId();
+ final Http2StreamChannelImpl channel = new Http2StreamChannelImpl(
+ promisedStreamId,
+ localConfig.getInitialWindowSize(),
+ remoteConfig.getInitialWindowSize());
+ final Http2StreamHandler streamHandler = new ServerPushHttp2StreamHandler(channel, pushProducer);
+ final Http2Stream stream = new Http2Stream(channel, streamHandler, false);
+ streamMap.put(promisedStreamId, stream);
+
+ outputLock.lock();
+ try {
+ if (localEndStream) {
+ stream.releaseResources();
+ return;
+ }
+ commitPushPromise(id, promisedStreamId, headers);
+ } finally {
+ outputLock.unlock();
+ }
+ }
+
+ @Override
+ public void update(final int increment) throws IOException {
+ if (remoteEndStream) {
+ return;
+ }
+ updateInputCapacity(id, inputWindow, increment);
+ }
+
+ @Override
+ public int write(final ByteBuffer payload) throws IOException {
+ outputLock.lock();
+ try {
+ if (localEndStream) {
+ return 0;
+ }
+ return streamData(id, outputWindow, payload);
+ } finally {
+ outputLock.unlock();
+ }
+ }
+
+ @Override
+ public void endStream(final List<Header> trailers) throws IOException {
+ outputLock.lock();
+ try {
+ if (localEndStream) {
+ return;
+ }
+ localEndStream = true;
+ final RawFrame frame = frameFactory.createData(id, null, true);
+ commitFrameInternal(frame);
+ } finally {
+ outputLock.unlock();
+ }
+ }
+
+ @Override
+ public void endStream() throws IOException {
+ endStream(null);
+ }
+
+ @Override
+ public void requestOutput() {
+ requestSessionOutput();
+ }
+
+ boolean isRemoteClosed() {
+ return remoteEndStream;
+ }
+
+ void setRemoteEndStream() {
+ remoteEndStream = true;
+ }
+
+ boolean isLocalClosed() {
+ return localEndStream;
+ }
+
+ boolean isClosed() {
+ return remoteEndStream && localEndStream;
+ }
+
+ void close() {
+ localEndStream = true;
+ remoteEndStream = true;
+ }
+
+ void localReset() throws IOException {
+ deadline = System.currentTimeMillis() + LINGER_TIME;
+ outputLock.lock();
+ try {
+ close();
+ final RawFrame resetStream = frameFactory.createResetStream(id, H2Error.REFUSED_STREAM);
+ commitFrameInternal(resetStream);
+ } finally {
+ outputLock.unlock();
+ }
+ }
+
+ long getDeadline() {
+ return deadline;
+ }
+
+ @Override
+ public String toString() {
+ return "[" +
+ "id=" + id +
+ ", inputWindow=" + inputWindow +
+ ", outputWindow=" + outputWindow +
+ ", remoteEndStream=" + remoteEndStream +
+ ", localEndStream=" + localEndStream +
+ ']';
+ }
+
+ }
+
+ private static class Http2Stream {
+
+ private final Http2StreamChannelImpl channel;
+ private final Http2StreamHandler handler;
+ private final boolean remoteInitiated;
+
+ private volatile boolean resetLocally;
+
+ private Http2Stream(
+ final Http2StreamChannelImpl channel,
+ final Http2StreamHandler handler,
+ final boolean remoteInitiated) {
+ this.channel = channel;
+ this.handler = handler;
+ this.remoteInitiated = remoteInitiated;
+ }
+
+ int getId() {
+ return channel.getId();
+ }
+
+ public boolean isRemoteInitiated() {
+ return remoteInitiated;
+ }
+
+ AtomicInteger getOutputWindow() {
+ return channel.getOutputWindow();
+ }
+
+ AtomicInteger getInputWindow() {
+ return channel.getInputWindow();
+ }
+
+ boolean isTerminated() {
+ return channel.isClosed() && channel.getDeadline() < System.currentTimeMillis();
+ }
+
+ boolean isRemoteClosed() {
+ return channel.isRemoteClosed();
+ }
+
+ boolean isLocalClosed() {
+ return channel.isLocalClosed();
+ }
+
+ void setRemoteEndStream() {
+ channel.setRemoteEndStream();
+ }
+
+ void consumePromise(final List<Header> headers) throws HttpException, IOException {
+ handler.consumePromise(headers);
+ }
+
+ void consumeHeader(final List<Header> headers) throws HttpException, IOException {
+ handler.consumeHeader(headers, channel.isRemoteClosed());
+ }
+
+ void consumeData(final ByteBuffer src) throws HttpException, IOException {
+ handler.consumeData(src, channel.isRemoteClosed());
+ }
+
+ boolean isOutputReady() {
+ return handler.isOutputReady();
+ }
+
+ void produceOutput() throws HttpException, IOException {
+ handler.produceOutput();
+ }
+
+ void produceInputCapacityUpdate() throws IOException {
+ handler.updateInputCapacity();
+ }
+
+ void reset(final Exception cause) {
+ channel.close();
+ handler.failed(cause);
+ }
+
+ void localReset(final Exception cause) throws IOException {
+ resetLocally = true;
+ channel.localReset();
+ handler.failed(cause);
+ }
+
+ public boolean isResetLocally() {
+ return resetLocally;
+ }
+
+ void cancel() {
+ channel.close();
+ handler.cancel();
+ }
+
+ void releaseResources() {
+ handler.releaseResources();
+ channel.requestOutput();
+ }
+
+ @Override
+ public String toString() {
+ return channel.toString();
+ }
+
+ }
+
+}
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractHttp2StreamMultiplexer.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractHttp2StreamMultiplexer.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractHttp2StreamMultiplexer.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Copied: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2IOEventHandler.java (from r1761317, httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/frame/FrameConsts.java)
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2IOEventHandler.java?p2=httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2IOEventHandler.java&p1=httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/frame/FrameConsts.java&r1=1761317&r2=1761318&rev=1761318&view=diff
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/frame/FrameConsts.java (original)
+++ httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2IOEventHandler.java Sun Sep 18 11:29:30 2016
@@ -24,17 +24,26 @@
* <http://www.apache.org/>.
*
*/
-package org.apache.hc.core5.http2.frame;
-public final class FrameConsts {
+package org.apache.hc.core5.http2.impl.nio;
- private FrameConsts() {
- // Do not allow utility class to be instantiated.
+import org.apache.hc.core5.annotation.Contract;
+import org.apache.hc.core5.annotation.ThreadingBehavior;
+
+/**
+ * {@link org.apache.hc.core5.reactor.IOEventHandler} that implements client side HTTP/2 messaging protocol.
+ *
+ * @since 5.0
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE_CONDITIONAL)
+public class ClientHttp2IOEventHandler extends AbstractHttp2IOEventHandler {
+
+ public ClientHttp2IOEventHandler(final ClientHttp2StreamMultiplexer streamMultiplexer, final HttpErrorListener errorListener) {
+ super(streamMultiplexer, errorListener);
}
- public final static int HEAD_LEN = 9;
- public final static int MAX_PADDING = 255;
- public final static int MIN_FRAME_SIZE = 2 ^ 14;
- public final static int MAX_FRAME_SIZE = 2 ^ 24 - 1;
+ public ClientHttp2IOEventHandler(final ClientHttp2StreamMultiplexer streamMultiplexer) {
+ this(streamMultiplexer, null);
+ }
-};
+}
Added: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamHandler.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamHandler.java?rev=1761318&view=auto
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamHandler.java (added)
+++ httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamHandler.java Sun Sep 18 11:29:30 2016
@@ -0,0 +1,247 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.http2.impl.nio;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.ProtocolException;
+import org.apache.hc.core5.http2.H2ConnectionException;
+import org.apache.hc.core5.http2.H2Error;
+import org.apache.hc.core5.http2.impl.DefaultH2RequestConverter;
+import org.apache.hc.core5.http2.impl.DefaultH2ResponseConverter;
+import org.apache.hc.core5.http2.nio.AsyncPushProducer;
+import org.apache.hc.core5.http2.nio.AsyncRequestProducer;
+import org.apache.hc.core5.http2.nio.AsyncResponseConsumer;
+
+class ClientHttp2StreamHandler<T> implements Http2StreamHandler {
+
+ private final AsyncRequestProducer requestProducer;
+ private final AsyncResponseConsumer<T> responseConsumer;
+ private final FutureCallback<T> resultCallback;
+ private final AtomicBoolean done;
+
+ private volatile MessageState requestState;
+ private volatile MessageState responseState;
+ private volatile Http2StreamChannel internalOutputChannel;
+
+ ClientHttp2StreamHandler(
+ final Http2StreamChannel outputChannel,
+ final AsyncRequestProducer requestProducer,
+ final AsyncResponseConsumer<T> responseConsumer,
+ final FutureCallback<T> resultCallback) {
+ this.internalOutputChannel = new Http2StreamChannel() {
+
+ @Override
+ public void submit(final List<Header> headers, final boolean endStream) throws HttpException, IOException {
+ outputChannel.submit(headers, endStream);
+ requestState = endStream ? MessageState.COMPLETE : MessageState.BODY;
+ }
+
+ @Override
+ public void push(final List<Header> headers, final AsyncPushProducer pushProducer) throws IOException {
+ throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Push not allowed");
+ }
+
+ @Override
+ public void update(final int increment) throws IOException {
+ outputChannel.update(increment);
+ }
+
+ @Override
+ public void requestOutput() {
+ outputChannel.requestOutput();
+ }
+
+ @Override
+ public int write(final ByteBuffer src) throws IOException {
+ return outputChannel.write(src);
+ }
+
+ @Override
+ public void endStream(final List<Header> trailers) throws IOException {
+ outputChannel.endStream(trailers);
+ requestState = MessageState.COMPLETE;
+ }
+
+ @Override
+ public void endStream() throws IOException {
+ outputChannel.endStream();
+ requestState = MessageState.COMPLETE;
+ }
+
+ };
+ this.requestProducer = requestProducer;
+ this.responseConsumer = responseConsumer;
+ this.resultCallback = resultCallback;
+ this.done = new AtomicBoolean(false);
+ this.requestState = MessageState.HEADERS;
+ this.responseState = MessageState.HEADERS;
+ }
+
+ @Override
+ public boolean isOutputReady() {
+ switch (requestState) {
+ case HEADERS:
+ return true;
+ case BODY:
+ return requestProducer.available() > 0;
+ default:
+ return false;
+ }
+ }
+
+ @Override
+ public void produceOutput() throws HttpException, IOException {
+ switch (requestState) {
+ case HEADERS:
+ final HttpRequest request = requestProducer.produceRequest();
+ final List<Header> headers = DefaultH2RequestConverter.INSTANCE.convert(request);
+ final boolean endStream = !requestProducer.isEnclosingEntity();
+ internalOutputChannel.submit(headers, endStream);
+ if (!endStream) {
+ requestProducer.dataStart(internalOutputChannel);
+ }
+ break;
+ case BODY:
+ requestProducer.produce(internalOutputChannel);
+ break;
+ }
+ }
+
+ @Override
+ public void consumePromise(final List<Header> headers) throws HttpException, IOException {
+ throw new ProtocolException("Unexpected message promise");
+ }
+
+ @Override
+ public void consumeHeader(final List<Header> headers, final boolean endStream) throws HttpException, IOException {
+ if (done.get() || responseState != MessageState.HEADERS) {
+ throw new ProtocolException("Unexpected message headers");
+ }
+ final HttpResponse response = DefaultH2ResponseConverter.INSTANCE.convert(headers);
+ responseConsumer.consumeResponse(response, new FutureCallback<T>() {
+
+ @Override
+ public void completed(final T result) {
+ if (resultCallback != null) {
+ resultCallback.completed(result);
+ }
+ }
+
+ @Override
+ public void failed(final Exception ex) {
+ if (resultCallback != null) {
+ resultCallback.failed(ex);
+ }
+ }
+
+ @Override
+ public void cancelled() {
+ if (resultCallback != null) {
+ resultCallback.cancelled();
+ }
+ }
+
+ });
+ if (endStream) {
+ responseState = MessageState.COMPLETE;
+ responseConsumer.streamEnd(null);
+ } else {
+ responseState = MessageState.BODY;
+ }
+ }
+
+ @Override
+ public void updateInputCapacity() throws IOException {
+ responseConsumer.updateCapacity(internalOutputChannel);
+ }
+
+ @Override
+ public void consumeData(final ByteBuffer src, final boolean endStream) throws HttpException, IOException {
+ if (done.get() || responseState != MessageState.BODY) {
+ throw new ProtocolException("Unexpected message data");
+ }
+ if (src != null) {
+ responseConsumer.consume(src);
+ }
+ if (endStream) {
+ responseState = MessageState.COMPLETE;
+ responseConsumer.streamEnd(null);
+ }
+ }
+
+ @Override
+ public void failed(final Exception cause) {
+ try {
+ if (resultCallback != null) {
+ resultCallback.failed(cause);
+ }
+ requestProducer.failed(cause);
+ } finally {
+ releaseResources();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ try {
+ if (resultCallback != null) {
+ resultCallback.cancelled();
+ }
+ } finally {
+ releaseResources();
+ }
+ }
+
+ @Override
+ public void releaseResources() {
+ if (done.compareAndSet(false, true)) {
+ responseState = MessageState.COMPLETE;
+ responseConsumer.releaseResources();
+ requestState = MessageState.COMPLETE;
+ requestProducer.releaseResources();
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "[" +
+ "requestState=" + requestState +
+ ", responseState=" + responseState +
+ ']';
+ }
+
+}
+
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamHandler.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamHandler.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamHandler.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamMultiplexer.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamMultiplexer.java?rev=1761318&view=auto
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamMultiplexer.java (added)
+++ httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamMultiplexer.java Sun Sep 18 11:29:30 2016
@@ -0,0 +1,81 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.http2.impl.nio;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+
+import org.apache.hc.core5.http2.config.H2Config;
+import org.apache.hc.core5.http2.frame.DefaultFrameFactory;
+import org.apache.hc.core5.http2.frame.FrameFactory;
+import org.apache.hc.core5.http2.frame.StreamIdGenerator;
+import org.apache.hc.core5.http2.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http2.nio.HandlerFactory;
+import org.apache.hc.core5.reactor.IOSession;
+
+/**
+ * Client side HTTP/2 stream multiplexer.
+ *
+ * @since 5.0
+ */
+public class ClientHttp2StreamMultiplexer extends AbstractHttp2StreamMultiplexer {
+
+ private final HandlerFactory<AsyncPushConsumer> pushHandlerFactory;
+
+ public ClientHttp2StreamMultiplexer(
+ final IOSession ioSession,
+ final FrameFactory frameFactory,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final Charset charset,
+ final H2Config h2Config,
+ final Http2StreamListener streamListener) {
+ super(Mode.CLIENT, ioSession, frameFactory, StreamIdGenerator.ODD, charset, h2Config, streamListener);
+ this.pushHandlerFactory = pushHandlerFactory;
+ }
+
+ public ClientHttp2StreamMultiplexer(
+ final IOSession ioSession,
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final Charset charset,
+ final H2Config h2Config) {
+ this(ioSession, DefaultFrameFactory.INSTANCE, pushHandlerFactory, charset, h2Config, null);
+ }
+
+ public ClientHttp2StreamMultiplexer(
+ final IOSession ioSession,
+ final Charset charset,
+ final H2Config h2Config) {
+ this(ioSession, null, charset, h2Config);
+ }
+
+ @Override
+ Http2StreamHandler createRemotelyInitiatedStream(final Http2StreamChannel channel) throws IOException {
+ return new ClientPushHttp2StreamHandler(channel, pushHandlerFactory);
+ }
+
+}
+
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamMultiplexer.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamMultiplexer.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttp2StreamMultiplexer.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java?rev=1761318&view=auto
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java (added)
+++ httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java Sun Sep 18 11:29:30 2016
@@ -0,0 +1,121 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.http2.impl.nio;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ByteChannel;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.hc.core5.annotation.Contract;
+import org.apache.hc.core5.annotation.ThreadingBehavior;
+import org.apache.hc.core5.http2.H2ConnectionException;
+import org.apache.hc.core5.http2.H2Error;
+import org.apache.hc.core5.http2.config.H2Config;
+import org.apache.hc.core5.http2.frame.DefaultFrameFactory;
+import org.apache.hc.core5.http2.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http2.nio.HandlerFactory;
+import org.apache.hc.core5.reactor.IOEventHandler;
+import org.apache.hc.core5.reactor.IOSession;
+
+/**
+ * @since 5.0
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE_CONDITIONAL)
+public class ClientHttpProtocolNegotiator implements IOEventHandler {
+
+ // PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n
+ final static byte[] PREFACE = new byte[] {
+ 0x50, 0x52, 0x49, 0x20, 0x2a, 0x20, 0x48, 0x54, 0x54, 0x50,
+ 0x2f, 0x32, 0x2e, 0x30, 0x0d, 0x0a, 0x0d, 0x0a, 0x53, 0x4d,
+ 0x0d, 0x0a, 0x0d, 0x0a};
+
+ private final Charset charset;
+ private final H2Config h2Config;
+ private final HandlerFactory<AsyncPushConsumer> pushHandlerFactory;
+ private final Http2StreamListener streamListener;
+ private final HttpErrorListener errorListener;
+
+ public ClientHttpProtocolNegotiator(
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final Charset charset,
+ final H2Config h2Config,
+ final Http2StreamListener streamListener,
+ final HttpErrorListener errorListener) {
+ this.pushHandlerFactory = pushHandlerFactory;
+ this.charset = charset != null ? charset : StandardCharsets.US_ASCII;
+ this.h2Config = h2Config != null ? h2Config : H2Config.DEFAULT;
+ this.streamListener = streamListener;
+ this.errorListener = errorListener;
+ }
+
+ protected ClientHttp2StreamMultiplexer createStreamMultiplexer(final IOSession ioSession) {
+ return new ClientHttp2StreamMultiplexer(ioSession, DefaultFrameFactory.INSTANCE,
+ pushHandlerFactory, charset, h2Config, streamListener);
+ }
+
+ @Override
+ public void connected(final IOSession ioSession) {
+ try {
+ final ByteChannel channel = ioSession.channel();
+ final ByteBuffer preface = ByteBuffer.wrap(PREFACE);
+ final int bytesWritten = channel.write(preface);
+ if (bytesWritten != PREFACE.length) {
+ throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "HTTP/2 preface failed");
+ }
+ } catch (IOException ex) {
+ ioSession.shutdown();
+ if (errorListener != null) {
+ errorListener.onError(ex);
+ }
+ return;
+ }
+ final ClientHttp2StreamMultiplexer streamMultiplexer = createStreamMultiplexer(ioSession);
+ final IOEventHandler newHandler = new ClientHttp2IOEventHandler(streamMultiplexer, errorListener);
+ newHandler.connected(ioSession);
+ ioSession.setHandler(newHandler);
+ }
+
+ @Override
+ public void inputReady(final IOSession session) {
+ }
+
+ @Override
+ public void outputReady(final IOSession session) {
+ }
+
+ @Override
+ public void timeout(final IOSession session) {
+ }
+
+ @Override
+ public void disconnected(final IOSession session) {
+ }
+
+}
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiator.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiatorFactory.java
URL: http://svn.apache.org/viewvc/httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiatorFactory.java?rev=1761318&view=auto
==============================================================================
--- httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiatorFactory.java (added)
+++ httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiatorFactory.java Sun Sep 18 11:29:30 2016
@@ -0,0 +1,84 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.core5.http2.impl.nio;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.hc.core5.annotation.Contract;
+import org.apache.hc.core5.annotation.ThreadingBehavior;
+import org.apache.hc.core5.http2.config.H2Config;
+import org.apache.hc.core5.http2.nio.AsyncPushConsumer;
+import org.apache.hc.core5.http2.nio.HandlerFactory;
+import org.apache.hc.core5.reactor.IOEventHandlerFactory;
+import org.apache.hc.core5.reactor.IOSession;
+
+/**
+ * @since 5.0
+ */
+@Contract(threading = ThreadingBehavior.IMMUTABLE)
+public class ClientHttpProtocolNegotiatorFactory implements IOEventHandlerFactory {
+
+ private final HandlerFactory<AsyncPushConsumer> pushHandlerFactory;
+ private final Charset charset;
+ private final H2Config h2Config;
+ private final Http2StreamListener streamListener;
+ private final HttpErrorListener errorListener;
+
+ public ClientHttpProtocolNegotiatorFactory(
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final Charset charset,
+ final H2Config h2Config,
+ final Http2StreamListener streamListener,
+ final HttpErrorListener errorListener) {
+ this.pushHandlerFactory = pushHandlerFactory;
+ this.charset = charset != null ? charset : StandardCharsets.US_ASCII;
+ this.h2Config = h2Config != null ? h2Config : H2Config.DEFAULT;
+ this.streamListener = streamListener;
+ this.errorListener = errorListener;
+ }
+
+ public ClientHttpProtocolNegotiatorFactory(
+ final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+ final Http2StreamListener streamListener,
+ final HttpErrorListener errorListener) {
+ this(pushHandlerFactory, null, null, streamListener, errorListener);
+ }
+
+ public ClientHttpProtocolNegotiatorFactory(
+ final Http2StreamListener streamListener,
+ final HttpErrorListener errorListener) {
+ this(null, streamListener, errorListener);
+ }
+
+ @Override
+ public ClientHttpProtocolNegotiator createHandler(final IOSession ioSession) {
+ return new ClientHttpProtocolNegotiator(pushHandlerFactory, charset, h2Config, streamListener, errorListener);
+ }
+
+}
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiatorFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiatorFactory.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: httpcomponents/httpcore/trunk/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/ClientHttpProtocolNegotiatorFactory.java
------------------------------------------------------------------------------
svn:mime-type = text/plain