@Test void testOnDataRead_endOfStreamExecutes() { testHandler = new Http2Handler(mockEncoder, MAX_PAYLOAD, NO_CORS); XrpcRequest fakeRequest = new XrpcRequest((Http2Headers) null, null, null, channel); testHandler.requests.put(STREAM_ID, fakeRequest); testHandler.handlers.put(STREAM_ID, OK_HANDLER); testHandler.onDataRead( mockContext, STREAM_ID, Unpooled.wrappedBuffer(new byte[] {0x20}), 0, true); assertEquals(0L, requestMeter.getCount()); verifyResponse( HttpResponseStatus.OK, ImmutableMap.of(), Optional.of(Unpooled.wrappedBuffer(new byte[] {0x20})), STREAM_ID); }
@Override public int onDataRead( ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) { XrpcRequest request = requests.get(streamId); int totalRead = request.addData(data); int processed = data.readableBytes() + padding; if (totalRead > maxPayloadBytes) { writeResponse( ctx, streamId, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, Unpooled.wrappedBuffer(XrpcConstants.PAYLOAD_EXCEEDED_RESPONSE)); ctx.flush(); ctx.close(); return processed; } if (endOfStream) { Optional<HttpResponse> corsResponse = corsHandler.inbound(request.h2Headers(), streamId); if (corsResponse.isPresent()) { writeResponse(ctx, streamId, corsResponse.get()); return processed; } Handler handler = handlers.get(streamId); try { HttpResponse response = handler.handle(request); writeResponse(ctx, streamId, response); return processed; } catch (IOException e) { log.error("Error in handling Route", e); ByteBuf buf = ctx.channel().alloc().directBuffer(); buf.writeBytes("Error executing endpoint".getBytes(XrpcConstants.DEFAULT_CHARSET)); writeResponse(ctx, streamId, HttpResponseStatus.INTERNAL_SERVER_ERROR, buf); } } return processed; }
Http2Handler extends Http2EventAdapter { @Override public int onDataRead( ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) { XrpcRequest request = requests.get(streamId); int totalRead = request.addData(data); int processed = data.readableBytes() + padding; if (totalRead > maxPayloadBytes) { writeResponse( ctx, streamId, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, Unpooled.wrappedBuffer(XrpcConstants.PAYLOAD_EXCEEDED_RESPONSE)); ctx.flush(); ctx.close(); return processed; } if (endOfStream) { Optional<HttpResponse> corsResponse = corsHandler.inbound(request.h2Headers(), streamId); if (corsResponse.isPresent()) { writeResponse(ctx, streamId, corsResponse.get()); return processed; } Handler handler = handlers.get(streamId); try { HttpResponse response = handler.handle(request); writeResponse(ctx, streamId, response); return processed; } catch (IOException e) { log.error("Error in handling Route", e); ByteBuf buf = ctx.channel().alloc().directBuffer(); buf.writeBytes("Error executing endpoint".getBytes(XrpcConstants.DEFAULT_CHARSET)); writeResponse(ctx, streamId, HttpResponseStatus.INTERNAL_SERVER_ERROR, buf); } } return processed; } }
Http2Handler extends Http2EventAdapter { @Override public int onDataRead( ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) { XrpcRequest request = requests.get(streamId); int totalRead = request.addData(data); int processed = data.readableBytes() + padding; if (totalRead > maxPayloadBytes) { writeResponse( ctx, streamId, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, Unpooled.wrappedBuffer(XrpcConstants.PAYLOAD_EXCEEDED_RESPONSE)); ctx.flush(); ctx.close(); return processed; } if (endOfStream) { Optional<HttpResponse> corsResponse = corsHandler.inbound(request.h2Headers(), streamId); if (corsResponse.isPresent()) { writeResponse(ctx, streamId, corsResponse.get()); return processed; } Handler handler = handlers.get(streamId); try { HttpResponse response = handler.handle(request); writeResponse(ctx, streamId, response); return processed; } catch (IOException e) { log.error("Error in handling Route", e); ByteBuf buf = ctx.channel().alloc().directBuffer(); buf.writeBytes("Error executing endpoint".getBytes(XrpcConstants.DEFAULT_CHARSET)); writeResponse(ctx, streamId, HttpResponseStatus.INTERNAL_SERVER_ERROR, buf); } } return processed; } Http2Handler(Http2ConnectionEncoder encoder, int maxPayloadBytes, Http2CorsHandler corsHandler); }
Http2Handler extends Http2EventAdapter { @Override public int onDataRead( ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) { XrpcRequest request = requests.get(streamId); int totalRead = request.addData(data); int processed = data.readableBytes() + padding; if (totalRead > maxPayloadBytes) { writeResponse( ctx, streamId, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, Unpooled.wrappedBuffer(XrpcConstants.PAYLOAD_EXCEEDED_RESPONSE)); ctx.flush(); ctx.close(); return processed; } if (endOfStream) { Optional<HttpResponse> corsResponse = corsHandler.inbound(request.h2Headers(), streamId); if (corsResponse.isPresent()) { writeResponse(ctx, streamId, corsResponse.get()); return processed; } Handler handler = handlers.get(streamId); try { HttpResponse response = handler.handle(request); writeResponse(ctx, streamId, response); return processed; } catch (IOException e) { log.error("Error in handling Route", e); ByteBuf buf = ctx.channel().alloc().directBuffer(); buf.writeBytes("Error executing endpoint".getBytes(XrpcConstants.DEFAULT_CHARSET)); writeResponse(ctx, streamId, HttpResponseStatus.INTERNAL_SERVER_ERROR, buf); } } return processed; } Http2Handler(Http2ConnectionEncoder encoder, int maxPayloadBytes, Http2CorsHandler corsHandler); @Override int onDataRead(
      ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream); @Override void onHeadersRead(
      ChannelHandlerContext ctx,
      int streamId,
      Http2Headers headers,
      int padding,
      boolean endOfStream); @Override void onHeadersRead(
      ChannelHandlerContext ctx,
      int streamId,
      Http2Headers headers,
      int streamDependency,
      short weight,
      boolean exclusive,
      int padding,
      boolean endOfStream); @Override void onStreamRemoved(Http2Stream stream); }
Http2Handler extends Http2EventAdapter { @Override public int onDataRead( ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream) { XrpcRequest request = requests.get(streamId); int totalRead = request.addData(data); int processed = data.readableBytes() + padding; if (totalRead > maxPayloadBytes) { writeResponse( ctx, streamId, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, Unpooled.wrappedBuffer(XrpcConstants.PAYLOAD_EXCEEDED_RESPONSE)); ctx.flush(); ctx.close(); return processed; } if (endOfStream) { Optional<HttpResponse> corsResponse = corsHandler.inbound(request.h2Headers(), streamId); if (corsResponse.isPresent()) { writeResponse(ctx, streamId, corsResponse.get()); return processed; } Handler handler = handlers.get(streamId); try { HttpResponse response = handler.handle(request); writeResponse(ctx, streamId, response); return processed; } catch (IOException e) { log.error("Error in handling Route", e); ByteBuf buf = ctx.channel().alloc().directBuffer(); buf.writeBytes("Error executing endpoint".getBytes(XrpcConstants.DEFAULT_CHARSET)); writeResponse(ctx, streamId, HttpResponseStatus.INTERNAL_SERVER_ERROR, buf); } } return processed; } Http2Handler(Http2ConnectionEncoder encoder, int maxPayloadBytes, Http2CorsHandler corsHandler); @Override int onDataRead(
      ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream); @Override void onHeadersRead(
      ChannelHandlerContext ctx,
      int streamId,
      Http2Headers headers,
      int padding,
      boolean endOfStream); @Override void onHeadersRead(
      ChannelHandlerContext ctx,
      int streamId,
      Http2Headers headers,
      int streamDependency,
      short weight,
      boolean exclusive,
      int padding,
      boolean endOfStream); @Override void onStreamRemoved(Http2Stream stream);  }