io.vertx.core.http.HttpServerResponse#endHandler ( )源码实例Demo

private void openAudioStream(HttpServerRequest request) {
  logger.info("New streamer");
  HttpServerResponse response = request.response()
    .putHeader("Content-Type", "audio/mpeg")
    .setChunked(true);
  streamers.add(response);
  response.endHandler(v -> {
    streamers.remove(response);
    logger.info("A streamer left");
  });
}
 

private void sse(HttpServerRequest request) {
  HttpServerResponse response = request.response();
  response
    .putHeader("Content-Type", "text/event-stream")
    .putHeader("Cache-Control", "no-cache")
    .setChunked(true);

  MessageConsumer<JsonObject> consumer = vertx.eventBus().consumer("sensor.updates");
  consumer.handler(msg -> {
    response.write("event: update\n");
    response.write("data: " + msg.body().encode() + "\n\n");
  });


  TimeoutStream ticks = vertx.periodicStream(1000);
  ticks.handler(id -> {
    vertx.eventBus().<JsonObject>request("sensor.average", "", reply -> {
      if (reply.succeeded()) {
        response.write("event: average\n");
        response.write("data: " + reply.result().body().encode() + "\n\n");
      }
    });
  });

  response.endHandler(v -> {
    consumer.unregister();
    ticks.cancel();
  });
}
 

@Override
public int read(final byte[] b, final int off, final int len) throws IOException {
    if (closed) {
        throw new IOException("Stream is closed");
    }
    readIntoBuffer();
    if (limit > 0 && exchange.request.bytesRead() > limit) {
        HttpServerResponse response = exchange.request.response();
        if (response.headWritten()) {
            //the response has been written, not much we can do
            exchange.request.connection().close();
            throw new IOException("Request too large");
        } else {
            response.setStatusCode(HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE.code());
            response.headers().add(HttpHeaderNames.CONNECTION, "close");
            response.endHandler(new Handler<Void>() {
                @Override
                public void handle(Void event) {
                    exchange.request.connection().close();
                }
            });
            response.end();
            throw new IOException("Request too large");
        }
    }
    if (finished) {
        return -1;
    }
    if (len == 0) {
        return 0;
    }
    ByteBuf buffer = pooled;
    int copied = Math.min(len, buffer.readableBytes());
    buffer.readBytes(b, off, copied);
    if (!buffer.isReadable()) {
        pooled.release();
        pooled = null;
    }
    return copied;
}