package io.netty.handler.codec.http2;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.Http2CodecUtil.SimpleChannelPromiseAggregator;
import io.netty.handler.codec.http2.Http2FrameWriter.Configuration;
import io.netty.handler.codec.http2.Http2HeadersEncoder.SensitivityDetector;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.UnstableApi;
import static io.netty.buffer.Unpooled.directBuffer;
import static io.netty.buffer.Unpooled.unreleasableBuffer;
import static io.netty.handler.codec.http2.Http2CodecUtil.CONTINUATION_FRAME_HEADER_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.DATA_FRAME_HEADER_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_MAX_FRAME_SIZE;
import static io.netty.handler.codec.http2.Http2CodecUtil.FRAME_HEADER_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.GO_AWAY_FRAME_HEADER_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.HEADERS_FRAME_HEADER_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.INT_FIELD_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_UNSIGNED_BYTE;
import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_UNSIGNED_INT;
import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_WEIGHT;
import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_WEIGHT;
import static io.netty.handler.codec.http2.Http2CodecUtil.PING_FRAME_PAYLOAD_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.PRIORITY_ENTRY_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.PRIORITY_FRAME_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.PUSH_PROMISE_FRAME_HEADER_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.RST_STREAM_FRAME_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.SETTING_ENTRY_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.WINDOW_UPDATE_FRAME_LENGTH;
import static io.netty.handler.codec.http2.Http2CodecUtil.isMaxFrameSizeValid;
import static io.netty.handler.codec.http2.Http2CodecUtil.verifyPadding;
import static io.netty.handler.codec.http2.Http2CodecUtil.writeFrameHeaderInternal;
import static io.netty.handler.codec.http2.Http2Error.FRAME_SIZE_ERROR;
import static io.netty.handler.codec.http2.Http2Exception.connectionError;
import static io.netty.handler.codec.http2.Http2FrameTypes.CONTINUATION;
import static io.netty.handler.codec.http2.Http2FrameTypes.DATA;
import static io.netty.handler.codec.http2.Http2FrameTypes.GO_AWAY;
import static io.netty.handler.codec.http2.Http2FrameTypes.HEADERS;
import static io.netty.handler.codec.http2.Http2FrameTypes.PING;
import static io.netty.handler.codec.http2.Http2FrameTypes.PRIORITY;
import static io.netty.handler.codec.http2.Http2FrameTypes.PUSH_PROMISE;
import static io.netty.handler.codec.http2.Http2FrameTypes.RST_STREAM;
import static io.netty.handler.codec.http2.Http2FrameTypes.SETTINGS;
import static io.netty.handler.codec.http2.Http2FrameTypes.WINDOW_UPDATE;
import static io.netty.util.internal.ObjectUtil.checkNotNull;
import static java.lang.Math.max;
import static java.lang.Math.min;
@UnstableApi
public class DefaultHttp2FrameWriter implements Http2FrameWriter, Http2FrameSizePolicy, Configuration {
private static final String STREAM_ID = "Stream ID";
private static final String STREAM_DEPENDENCY = "Stream Dependency";
private static final ByteBuf ZERO_BUFFER =
unreleasableBuffer(directBuffer(MAX_UNSIGNED_BYTE).writeZero(MAX_UNSIGNED_BYTE)).asReadOnly();
private final Http2HeadersEncoder headersEncoder;
private int maxFrameSize;
public DefaultHttp2FrameWriter() {
this(new DefaultHttp2HeadersEncoder());
}
public DefaultHttp2FrameWriter(SensitivityDetector headersSensitivityDetector) {
this(new DefaultHttp2HeadersEncoder(headersSensitivityDetector));
}
public DefaultHttp2FrameWriter(SensitivityDetector headersSensitivityDetector, boolean ignoreMaxHeaderListSize) {
this(new DefaultHttp2HeadersEncoder(headersSensitivityDetector, ignoreMaxHeaderListSize));
}
public DefaultHttp2FrameWriter(Http2HeadersEncoder headersEncoder) {
this.headersEncoder = headersEncoder;
maxFrameSize = DEFAULT_MAX_FRAME_SIZE;
}
@Override
public Configuration configuration() {
return this;
}
@Override
public Http2HeadersEncoder.Configuration headersConfiguration() {
return headersEncoder.configuration();
}
@Override
public Http2FrameSizePolicy frameSizePolicy() {
return this;
}
@Override
public void maxFrameSize(int max) throws Http2Exception {
if (!isMaxFrameSizeValid(max)) {
throw connectionError(FRAME_SIZE_ERROR, "Invalid MAX_FRAME_SIZE specified in sent settings: %d", max);
}
maxFrameSize = max;
}
@Override
public int maxFrameSize() {
return maxFrameSize;
}
@Override
public void close() { }
@Override
public ChannelFuture writeData(ChannelHandlerContext ctx, int streamId, ByteBuf data,
int padding, boolean endStream, ChannelPromise promise) {
final SimpleChannelPromiseAggregator promiseAggregator =
new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
ByteBuf frameHeader = null;
try {
verifyStreamId(streamId, STREAM_ID);
verifyPadding(padding);
int remainingData = data.readableBytes();
Http2Flags flags = new Http2Flags();
flags.endOfStream(false);
flags.paddingPresent(false);
if (remainingData > maxFrameSize) {
frameHeader = ctx.alloc().buffer(FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(frameHeader, maxFrameSize, DATA, flags, streamId);
do {
ctx.write(frameHeader.retainedSlice(), promiseAggregator.newPromise());
ctx.write(data.readRetainedSlice(maxFrameSize), promiseAggregator.newPromise());
remainingData -= maxFrameSize;
} while (remainingData > maxFrameSize);
}
if (padding == 0) {
if (frameHeader != null) {
frameHeader.release();
frameHeader = null;
}
ByteBuf frameHeader2 = ctx.alloc().buffer(FRAME_HEADER_LENGTH);
flags.endOfStream(endStream);
writeFrameHeaderInternal(frameHeader2, remainingData, DATA, flags, streamId);
ctx.write(frameHeader2, promiseAggregator.newPromise());
ByteBuf lastFrame = data.readSlice(remainingData);
data = null;
ctx.write(lastFrame, promiseAggregator.newPromise());
} else {
if (remainingData != maxFrameSize) {
if (frameHeader != null) {
frameHeader.release();
frameHeader = null;
}
} else {
remainingData -= maxFrameSize;
ByteBuf lastFrame;
if (frameHeader == null) {
lastFrame = ctx.alloc().buffer(FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(lastFrame, maxFrameSize, DATA, flags, streamId);
} else {
lastFrame = frameHeader.slice();
frameHeader = null;
}
ctx.write(lastFrame, promiseAggregator.newPromise());
lastFrame = data.readSlice(maxFrameSize);
data = null;
ctx.write(lastFrame, promiseAggregator.newPromise());
}
do {
int frameDataBytes = min(remainingData, maxFrameSize);
int framePaddingBytes = min(padding, max(0, (maxFrameSize - 1) - frameDataBytes));
padding -= framePaddingBytes;
remainingData -= frameDataBytes;
ByteBuf frameHeader2 = ctx.alloc().buffer(DATA_FRAME_HEADER_LENGTH);
flags.endOfStream(endStream && remainingData == 0 && padding == 0);
flags.paddingPresent(framePaddingBytes > 0);
writeFrameHeaderInternal(frameHeader2, framePaddingBytes + frameDataBytes, DATA, flags, streamId);
writePaddingLength(frameHeader2, framePaddingBytes);
ctx.write(frameHeader2, promiseAggregator.newPromise());
if (frameDataBytes != 0) {
if (remainingData == 0) {
ByteBuf lastFrame = data.readSlice(frameDataBytes);
data = null;
ctx.write(lastFrame, promiseAggregator.newPromise());
} else {
ctx.write(data.readRetainedSlice(frameDataBytes), promiseAggregator.newPromise());
}
}
if (paddingBytes(framePaddingBytes) > 0) {
ctx.write(ZERO_BUFFER.slice(0, paddingBytes(framePaddingBytes)),
promiseAggregator.newPromise());
}
} while (remainingData != 0 || padding != 0);
}
} catch (Throwable cause) {
if (frameHeader != null) {
frameHeader.release();
}
try {
if (data != null) {
data.release();
}
} finally {
promiseAggregator.setFailure(cause);
promiseAggregator.doneAllocatingPromises();
}
return promiseAggregator;
}
return promiseAggregator.doneAllocatingPromises();
}
@Override
public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId,
Http2Headers headers, int padding, boolean endStream, ChannelPromise promise) {
return writeHeadersInternal(ctx, streamId, headers, padding, endStream,
false, 0, (short) 0, false, promise);
}
@Override
public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId,
Http2Headers headers, int streamDependency, short weight, boolean exclusive,
int padding, boolean endStream, ChannelPromise promise) {
return writeHeadersInternal(ctx, streamId, headers, padding, endStream,
true, streamDependency, weight, exclusive, promise);
}
@Override
public ChannelFuture writePriority(ChannelHandlerContext ctx, int streamId,
int streamDependency, short weight, boolean exclusive, ChannelPromise promise) {
try {
verifyStreamId(streamId, STREAM_ID);
verifyStreamId(streamDependency, STREAM_DEPENDENCY);
verifyWeight(weight);
ByteBuf buf = ctx.alloc().buffer(PRIORITY_FRAME_LENGTH);
writeFrameHeaderInternal(buf, PRIORITY_ENTRY_LENGTH, PRIORITY, new Http2Flags(), streamId);
buf.writeInt(exclusive ? (int) (0x80000000L | streamDependency) : streamDependency);
buf.writeByte(weight - 1);
return ctx.write(buf, promise);
} catch (Throwable t) {
return promise.setFailure(t);
}
}
@Override
public ChannelFuture writeRstStream(ChannelHandlerContext ctx, int streamId, long errorCode,
ChannelPromise promise) {
try {
verifyStreamId(streamId, STREAM_ID);
verifyErrorCode(errorCode);
ByteBuf buf = ctx.alloc().buffer(RST_STREAM_FRAME_LENGTH);
writeFrameHeaderInternal(buf, INT_FIELD_LENGTH, RST_STREAM, new Http2Flags(), streamId);
buf.writeInt((int) errorCode);
return ctx.write(buf, promise);
} catch (Throwable t) {
return promise.setFailure(t);
}
}
@Override
public ChannelFuture writeSettings(ChannelHandlerContext ctx, Http2Settings settings,
ChannelPromise promise) {
try {
checkNotNull(settings, "settings");
int payloadLength = SETTING_ENTRY_LENGTH * settings.size();
ByteBuf buf = ctx.alloc().buffer(FRAME_HEADER_LENGTH + settings.size() * SETTING_ENTRY_LENGTH);
writeFrameHeaderInternal(buf, payloadLength, SETTINGS, new Http2Flags(), 0);
for (Http2Settings.PrimitiveEntry<Long> entry : settings.entries()) {
buf.writeChar(entry.key());
buf.writeInt(entry.value().intValue());
}
return ctx.write(buf, promise);
} catch (Throwable t) {
return promise.setFailure(t);
}
}
@Override
public ChannelFuture writeSettingsAck(ChannelHandlerContext ctx, ChannelPromise promise) {
try {
ByteBuf buf = ctx.alloc().buffer(FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(buf, 0, SETTINGS, new Http2Flags().ack(true), 0);
return ctx.write(buf, promise);
} catch (Throwable t) {
return promise.setFailure(t);
}
}
@Override
public ChannelFuture writePing(ChannelHandlerContext ctx, boolean ack, long data, ChannelPromise promise) {
Http2Flags flags = ack ? new Http2Flags().ack(true) : new Http2Flags();
ByteBuf buf = ctx.alloc().buffer(FRAME_HEADER_LENGTH + PING_FRAME_PAYLOAD_LENGTH);
writeFrameHeaderInternal(buf, PING_FRAME_PAYLOAD_LENGTH, PING, flags, 0);
buf.writeLong(data);
return ctx.write(buf, promise);
}
@Override
public ChannelFuture writePushPromise(ChannelHandlerContext ctx, int streamId,
int promisedStreamId, Http2Headers headers, int padding, ChannelPromise promise) {
ByteBuf headerBlock = null;
SimpleChannelPromiseAggregator promiseAggregator =
new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
try {
verifyStreamId(streamId, STREAM_ID);
verifyStreamId(promisedStreamId, "Promised Stream ID");
verifyPadding(padding);
headerBlock = ctx.alloc().buffer();
headersEncoder.encodeHeaders(streamId, headers, headerBlock);
Http2Flags flags = new Http2Flags().paddingPresent(padding > 0);
int nonFragmentLength = INT_FIELD_LENGTH + padding;
int maxFragmentLength = maxFrameSize - nonFragmentLength;
ByteBuf fragment = headerBlock.readRetainedSlice(min(headerBlock.readableBytes(), maxFragmentLength));
flags.endOfHeaders(!headerBlock.isReadable());
int payloadLength = fragment.readableBytes() + nonFragmentLength;
ByteBuf buf = ctx.alloc().buffer(PUSH_PROMISE_FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(buf, payloadLength, PUSH_PROMISE, flags, streamId);
writePaddingLength(buf, padding);
buf.writeInt(promisedStreamId);
ctx.write(buf, promiseAggregator.newPromise());
ctx.write(fragment, promiseAggregator.newPromise());
if (paddingBytes(padding) > 0) {
ctx.write(ZERO_BUFFER.slice(0, paddingBytes(padding)), promiseAggregator.newPromise());
}
if (!flags.endOfHeaders()) {
writeContinuationFrames(ctx, streamId, headerBlock, padding, promiseAggregator);
}
} catch (Http2Exception e) {
promiseAggregator.setFailure(e);
} catch (Throwable t) {
promiseAggregator.setFailure(t);
promiseAggregator.doneAllocatingPromises();
PlatformDependent.throwException(t);
} finally {
if (headerBlock != null) {
headerBlock.release();
}
}
return promiseAggregator.doneAllocatingPromises();
}
@Override
public ChannelFuture writeGoAway(ChannelHandlerContext ctx, int lastStreamId, long errorCode,
ByteBuf debugData, ChannelPromise promise) {
SimpleChannelPromiseAggregator promiseAggregator =
new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
try {
verifyStreamOrConnectionId(lastStreamId, "Last Stream ID");
verifyErrorCode(errorCode);
int payloadLength = 8 + debugData.readableBytes();
ByteBuf buf = ctx.alloc().buffer(GO_AWAY_FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(buf, payloadLength, GO_AWAY, new Http2Flags(), 0);
buf.writeInt(lastStreamId);
buf.writeInt((int) errorCode);
ctx.write(buf, promiseAggregator.newPromise());
} catch (Throwable t) {
try {
debugData.release();
} finally {
promiseAggregator.setFailure(t);
promiseAggregator.doneAllocatingPromises();
}
return promiseAggregator;
}
try {
ctx.write(debugData, promiseAggregator.newPromise());
} catch (Throwable t) {
promiseAggregator.setFailure(t);
}
return promiseAggregator.doneAllocatingPromises();
}
@Override
public ChannelFuture writeWindowUpdate(ChannelHandlerContext ctx, int streamId,
int windowSizeIncrement, ChannelPromise promise) {
try {
verifyStreamOrConnectionId(streamId, STREAM_ID);
verifyWindowSizeIncrement(windowSizeIncrement);
ByteBuf buf = ctx.alloc().buffer(WINDOW_UPDATE_FRAME_LENGTH);
writeFrameHeaderInternal(buf, INT_FIELD_LENGTH, WINDOW_UPDATE, new Http2Flags(), streamId);
buf.writeInt(windowSizeIncrement);
return ctx.write(buf, promise);
} catch (Throwable t) {
return promise.setFailure(t);
}
}
@Override
public ChannelFuture writeFrame(ChannelHandlerContext ctx, byte frameType, int streamId,
Http2Flags flags, ByteBuf payload, ChannelPromise promise) {
SimpleChannelPromiseAggregator promiseAggregator =
new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
try {
verifyStreamOrConnectionId(streamId, STREAM_ID);
ByteBuf buf = ctx.alloc().buffer(FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(buf, payload.readableBytes(), frameType, flags, streamId);
ctx.write(buf, promiseAggregator.newPromise());
} catch (Throwable t) {
try {
payload.release();
} finally {
promiseAggregator.setFailure(t);
promiseAggregator.doneAllocatingPromises();
}
return promiseAggregator;
}
try {
ctx.write(payload, promiseAggregator.newPromise());
} catch (Throwable t) {
promiseAggregator.setFailure(t);
}
return promiseAggregator.doneAllocatingPromises();
}
private ChannelFuture writeHeadersInternal(ChannelHandlerContext ctx,
int streamId, Http2Headers headers, int padding, boolean endStream,
boolean hasPriority, int streamDependency, short weight, boolean exclusive, ChannelPromise promise) {
ByteBuf headerBlock = null;
SimpleChannelPromiseAggregator promiseAggregator =
new SimpleChannelPromiseAggregator(promise, ctx.channel(), ctx.executor());
try {
verifyStreamId(streamId, STREAM_ID);
if (hasPriority) {
verifyStreamOrConnectionId(streamDependency, STREAM_DEPENDENCY);
verifyPadding(padding);
verifyWeight(weight);
}
headerBlock = ctx.alloc().buffer();
headersEncoder.encodeHeaders(streamId, headers, headerBlock);
Http2Flags flags =
new Http2Flags().endOfStream(endStream).priorityPresent(hasPriority).paddingPresent(padding > 0);
int nonFragmentBytes = padding + flags.getNumPriorityBytes();
int maxFragmentLength = maxFrameSize - nonFragmentBytes;
ByteBuf fragment = headerBlock.readRetainedSlice(min(headerBlock.readableBytes(), maxFragmentLength));
flags.endOfHeaders(!headerBlock.isReadable());
int payloadLength = fragment.readableBytes() + nonFragmentBytes;
ByteBuf buf = ctx.alloc().buffer(HEADERS_FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(buf, payloadLength, HEADERS, flags, streamId);
writePaddingLength(buf, padding);
if (hasPriority) {
buf.writeInt(exclusive ? (int) (0x80000000L | streamDependency) : streamDependency);
buf.writeByte(weight - 1);
}
ctx.write(buf, promiseAggregator.newPromise());
ctx.write(fragment, promiseAggregator.newPromise());
if (paddingBytes(padding) > 0) {
ctx.write(ZERO_BUFFER.slice(0, paddingBytes(padding)), promiseAggregator.newPromise());
}
if (!flags.endOfHeaders()) {
writeContinuationFrames(ctx, streamId, headerBlock, padding, promiseAggregator);
}
} catch (Http2Exception e) {
promiseAggregator.setFailure(e);
} catch (Throwable t) {
promiseAggregator.setFailure(t);
promiseAggregator.doneAllocatingPromises();
PlatformDependent.throwException(t);
} finally {
if (headerBlock != null) {
headerBlock.release();
}
}
return promiseAggregator.doneAllocatingPromises();
}
private ChannelFuture writeContinuationFrames(ChannelHandlerContext ctx, int streamId,
ByteBuf headerBlock, int padding, SimpleChannelPromiseAggregator promiseAggregator) {
Http2Flags flags = new Http2Flags().paddingPresent(padding > 0);
int maxFragmentLength = maxFrameSize - padding;
if (maxFragmentLength <= 0) {
return promiseAggregator.setFailure(new IllegalArgumentException(
"Padding [" + padding + "] is too large for max frame size [" + maxFrameSize + "]"));
}
if (headerBlock.isReadable()) {
int fragmentReadableBytes = min(headerBlock.readableBytes(), maxFragmentLength);
int payloadLength = fragmentReadableBytes + padding;
ByteBuf buf = ctx.alloc().buffer(CONTINUATION_FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(buf, payloadLength, CONTINUATION, flags, streamId);
writePaddingLength(buf, padding);
do {
fragmentReadableBytes = min(headerBlock.readableBytes(), maxFragmentLength);
ByteBuf fragment = headerBlock.readRetainedSlice(fragmentReadableBytes);
payloadLength = fragmentReadableBytes + padding;
if (headerBlock.isReadable()) {
ctx.write(buf.retain(), promiseAggregator.newPromise());
} else {
flags = flags.endOfHeaders(true);
buf.release();
buf = ctx.alloc().buffer(CONTINUATION_FRAME_HEADER_LENGTH);
writeFrameHeaderInternal(buf, payloadLength, CONTINUATION, flags, streamId);
writePaddingLength(buf, padding);
ctx.write(buf, promiseAggregator.newPromise());
}
ctx.write(fragment, promiseAggregator.newPromise());
if (paddingBytes(padding) > 0) {
ctx.write(ZERO_BUFFER.slice(0, paddingBytes(padding)), promiseAggregator.newPromise());
}
} while(headerBlock.isReadable());
}
return promiseAggregator;
}
private static int paddingBytes(int padding) {
return padding - 1;
}
private static void writePaddingLength(ByteBuf buf, int padding) {
if (padding > 0) {
buf.writeByte(padding - 1);
}
}
private static void verifyStreamId(int streamId, String argumentName) {
if (streamId <= 0) {
throw new IllegalArgumentException(argumentName + " must be > 0");
}
}
private static void verifyStreamOrConnectionId(int streamId, String argumentName) {
if (streamId < 0) {
throw new IllegalArgumentException(argumentName + " must be >= 0");
}
}
private static void verifyWeight(short weight) {
if (weight < MIN_WEIGHT || weight > MAX_WEIGHT) {
throw new IllegalArgumentException("Invalid weight: " + weight);
}
}
private static void verifyErrorCode(long errorCode) {
if (errorCode < 0 || errorCode > MAX_UNSIGNED_INT) {
throw new IllegalArgumentException("Invalid errorCode: " + errorCode);
}
}
private static void verifyWindowSizeIncrement(int windowSizeIncrement) {
if (windowSizeIncrement < 0) {
throw new IllegalArgumentException("WindowSizeIncrement must be >= 0");
}
}
private static void verifyPingPayload(ByteBuf data) {
if (data == null || data.readableBytes() != PING_FRAME_PAYLOAD_LENGTH) {
throw new IllegalArgumentException("Opaque data must be " + PING_FRAME_PAYLOAD_LENGTH + " bytes");
}
}
}