@DebugMetadata(f = "HttpVerticle.kt", l = {31, 32}, i = {}, s = {}, n = {}, m = "invokeSuspend", c = "org.wudan.vertx.HttpVerticle$start$2$1")
@Metadata(mv = {1, 6, 0}, k = 3, xi = 48, d1 = {"\000\n\n\000\n\002\020\002\n\002\030\002\020\000\032\0020\001*\0020\002HŠ@"}, d2 = {"<anonymous>", "", "Lkotlinx/coroutines/CoroutineScope;"})
final class null extends SuspendLambda implements Function2<CoroutineScope, Continuation<? super Unit>, Object> {
  int label;

  null(RoutingContext $cont, Continuation $completion) {
    super(2, $completion);
  }

  @Nullable
  public final Object invokeSuspend(@NotNull Object $result) {
    Object object = IntrinsicsKt.getCOROUTINE_SUSPENDED();
    switch (this.label) {
      case 0:
        ResultKt.throwOnFailure(SYNTHETIC_LOCAL_VARIABLE_1);
        Intrinsics.checkNotNullExpressionValue(Future.fromCompletionStage(HttpVerticle.this.getHealthCheckFuture2()), "fromCompletionStage(getHealthCheckFuture2())");
        this.label = 1;
        if (VertxCoroutineKt.await(Future.fromCompletionStage(HttpVerticle.this.getHealthCheckFuture2()), (Continuation)this) == object)
          return object;
        (Response)VertxCoroutineKt.await(Future.fromCompletionStage(HttpVerticle.this.getHealthCheckFuture2()), (Continuation)this);
        Intrinsics.checkNotNullExpressionValue(this.$cont.response().end("Ok!"), "cont.response().end(\"Ok!\")");
        this.label = 2;
        if (VertxCoroutineKt.await(this.$cont.response().end("Ok!"), (Continuation)this) == object)
          return object;
        VertxCoroutineKt.await(this.$cont.response().end("Ok!"), (Continuation)this);
        return Unit.INSTANCE;
      case 1:
        ResultKt.throwOnFailure(SYNTHETIC_LOCAL_VARIABLE_1);
        (Response)SYNTHETIC_LOCAL_VARIABLE_1;
        Intrinsics.checkNotNullExpressionValue(this.$cont.response().end("Ok!"), "cont.response().end(\"Ok!\")");
        this.label = 2;
        if (VertxCoroutineKt.await(this.$cont.response().end("Ok!"), (Continuation)this) == object)
          return object;
        VertxCoroutineKt.await(this.$cont.response().end("Ok!"), (Continuation)this);
        return Unit.INSTANCE;
      case 2:
        ResultKt.throwOnFailure(SYNTHETIC_LOCAL_VARIABLE_1);
        return Unit.INSTANCE;
    }
    throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
  }

  @NotNull
  public final Continuation<Unit> create(@Nullable Object value, @NotNull Continuation<? super null> $completion) {
    return (Continuation<Unit>)new Object(HttpVerticle.this, this.$cont, $completion);
  }

  @Nullable
  public final Object invoke(@NotNull CoroutineScope p1, @Nullable Continuation<?> p2) {
    return ((null)create(p1, p2)).invokeSuspend(Unit.INSTANCE);
  }
}

suspend fun <T> Future<T>.await(): T = when {
  succeeded() -> result()
  failed() -> throw cause()
  else -> suspendCancellableCoroutine { cont: CancellableContinuation<T> ->
    onComplete { asyncResult ->
      if (asyncResult.succeeded()) cont.resume(asyncResult.result() as T)
      else cont.resumeWithException(asyncResult.cause())
    }
  }
}

class HttpVerticle : CoroutineVerticle() {

    private val acyncClient : AsyncHttpClient = Dsl.asyncHttpClient(Dsl.config()
            // I/O线程限制不要太大
            .setIoThreadsCount(8)
            .setProxyServer(Dsl.proxyServer("192.168.1.4", 9090)))

    fun getHealthCheckFuture2() : CompletableFuture<Response?>? {
        return acyncClient.prepareGet("http://192.168.1.4:9090/healthcheck")
                .execute()
                .toCompletableFuture()
    }

    override suspend fun start() {
        var router = Router.router(vertx)
        router.route().handler { cont ->
            cont.request().pause()
            launch {
                val response = Future.fromCompletionStage(getHealthCheckFuture2()).await()
                cont.response().end("Ok!").await()
            }
        }
        vertx.createHttpServer().requestHandler(router).listen(8080).await()
        println("Server Starting!")
    }
}

suspend fun main() {
    var vertxOption = VertxOptions()
    vertxOption.maxEventLoopExecuteTime = TimeUnit.MILLISECONDS.toNanos(100)
    vertxOption.eventLoopPoolSize = 1 // 单线程运行

    var options = DeploymentOptions()
    options.instances = 1 // 单Verticle运行
    var deployVerticle = Vertx.vertx(vertxOption).deployVerticle("org.wudan.vertx.HttpVerticle", options).await()
}

EventExecutorGroup eventLoopGroup =  new DefaultEventExecutorGroup(8, new DefaultThreadFactory("wudan-group", true));

ClientResources clientResources = DefaultClientResources
    .builder()
    .eventExecutorGroup(eventLoopGroup)
    .build();
RedisClient redisClient = RedisClient.create(clientResources, "redis://localhost:6379");
StatefulRedisConnection<String, String> connection = redisClient.connect();
RedisReactiveCommands<String, String> reactiveCommand = connection.reactive();

reactiveCommand.get("wudan").subscribe(s -> {
    System.out.println(Thread.currentThread().getName() + " " + s);
});

Thread.sleep(1000L);

WeakReference<Object> weak = new WeakReference<>(new Object());
System.out.println(weak.get());// 返回object
System.gc();
System.out.println(weak.get());// 返回null

// -Xmx6m
SoftReference<int[]> soft = new SoftReference<>(new int[512 * 1024]);
System.out.println(soft.get());// 返回数组，因为内存还够
int[] arr2 = new int[512 * 1024]; // 内存不够了，会gc
arr2[0] = 1;
System.out.println(soft.get());// 返回null，因为内存不够了

ReferenceQueue referenceQueue = new ReferenceQueue();
PhantomReference<Object> soft = new PhantomReference<>(new Object(), referenceQueue);
System.out.println(soft.get());// 返回null
System.out.println(referenceQueue.poll());// poll不出来，因为没有gc
System.gc();
Thread.sleep(5000);
Reference poll = referenceQueue.poll();// 可以poll出来，因为gc了
System.out.println(poll);
if (poll != null) {
    System.out.println(poll.get());
}

private static void allocateDirect() {
    ByteBuffer byteBuffer = ByteBuffer.allocateDirect(1024);
    byteBuffer.put(0, (byte) 0x55);
    System.out.println(byteBuffer.get(0));
}
public static void main(String[] args) throws InterruptedException {
    allocateDirect();
    System.gc();
    Thread.sleep(5000);
    System.exit(0);
}

// Cleaner是PhantomReference的实现类，虚引用在引用队列中被poll出来之后，也是get不出来东西的，需要根据需要把一些数据放到PhantomReference中或者存储一个Reference到这些数据的一个对应关系map，才能在poll出来之后做一些实际的操作
cleaner = Cleaner.create(this, new Deallocator(base, size, cap));

/* List of References waiting to be enqueued.  The collector adds
 * References to this list, while the Reference-handler thread removes
 * them.  This list is protected by the above lock object. The
 * list uses the discovered field to link its elements.
 */
private static Reference<Object> pending = null;

public class NettyHttpServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(new DefaultThreadFactory("boss"));
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(4, new DefaultThreadFactory("worker"));

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_BACKLOG, 120)
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            channel.pipeline().addLast(new IdleStateHandler(10, 10, 10, TimeUnit.MINUTES));
                            channel.pipeline().addLast(new HttpRequestDecoder());
                            channel.pipeline().addLast(new HttpResponseEncoder());
                            channel.pipeline().addLast(new HttpObjectAggregator(4 * 1024 * 1024));
                            channel.pipeline().addLast(new HttpServerHandler());
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture channelFuture = serverBootstrap.bind(8888).sync();
            channelFuture.channel().closeFuture().get();
        } catch (Exception e) {
            System.out.println(e);
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    private static final String FAVICON_ICO = "/favicon.ico";
    private static final AsciiString CONNECTION = AsciiString.cached("Connection");
    private static final AsciiString KEEP_ALIVE = AsciiString.cached("keep-alive");
    private static final AsciiString CONTENT_TYPE = AsciiString.cached("Content-Type");
    private static final AsciiString CONTENT_LENGTH = AsciiString.cached("Content-Length");

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest fullHttpRequest) {
        String uri = fullHttpRequest.uri();
        if (uri.equals(FAVICON_ICO)) {
            return;
        }
        Object result;
        FullHttpResponse response;
        try {
            byte[] responseBytes = "{\"status\":0}".getBytes();
            response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(responseBytes));
            response.headers().set(CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
            response.headers().setInt(CONTENT_LENGTH, response.content().readableBytes());
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
            byte[] responseBytes = "{\"status\":1}".getBytes();
            response = new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR, Unpooled.wrappedBuffer(responseBytes));
            response.headers().set(CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
        }
        boolean keepAlive = HttpUtil.isKeepAlive(fullHttpRequest);
        if (!keepAlive) {
            ctx.write(response).addListener(ChannelFutureListener.CLOSE);
        } else {
            response.headers().set(CONNECTION, KEEP_ALIVE);
            ctx.write(response);
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }
}

protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
    if (resetRequested) {
        resetNow();
    }

    switch (currentState) {
    case SKIP_CONTROL_CHARS:
        // Fall-through
    case READ_INITIAL: try {
        AppendableCharSequence line = lineParser.parse(buffer);
        if (line == null) {
            return;
        }
        String[] initialLine = splitInitialLine(line);
        if (initialLine.length < 3) {
            // Invalid initial line - ignore.
            currentState = State.SKIP_CONTROL_CHARS;
            return;
        }

        message = createMessage(initialLine);
        currentState = State.READ_HEADER;
        // fall-through
    } catch (Exception e) {
        out.add(invalidMessage(buffer, e));
        return;
    }
    case READ_HEADER: try {
        State nextState = readHeaders(buffer);
        if (nextState == null) {
            return;
        }
        currentState = nextState;
        switch (nextState) {
        case SKIP_CONTROL_CHARS:
            // fast-path
            // No content is expected.
            out.add(message);
            out.add(LastHttpContent.EMPTY_LAST_CONTENT);
            resetNow();
            return;
        case READ_CHUNK_SIZE:
            if (!chunkedSupported) {
                throw new IllegalArgumentException("Chunked messages not supported");
            }
            // Chunked encoding - generate HttpMessage first.  HttpChunks will follow.
            out.add(message);
            return;
        default:
            /**
             * <a href="https://tools.ietf.org/html/rfc7230#section-3.3.3">RFC 7230, 3.3.3</a> states that if a
             * request does not have either a transfer-encoding or a content-length header then the message body
             * length is 0. However for a response the body length is the number of octets received prior to the
             * server closing the connection. So we treat this as variable length chunked encoding.
             */
            long contentLength = contentLength();
            if (contentLength == 0 || contentLength == -1 && isDecodingRequest()) {
                out.add(message);
                out.add(LastHttpContent.EMPTY_LAST_CONTENT);
                resetNow();
                return;
            }

            assert nextState == State.READ_FIXED_LENGTH_CONTENT ||
                    nextState == State.READ_VARIABLE_LENGTH_CONTENT;

            out.add(message);

            if (nextState == State.READ_FIXED_LENGTH_CONTENT) {
                // chunkSize will be decreased as the READ_FIXED_LENGTH_CONTENT state reads data chunk by chunk.
                chunkSize = contentLength;
            }

            // We return here, this forces decode to be called again where we will decode the content
            return;
        }
    } catch (Exception e) {
        out.add(invalidMessage(buffer, e));
        return;
    }
    case READ_VARIABLE_LENGTH_CONTENT: {
        // Keep reading data as a chunk until the end of connection is reached.
        int toRead = Math.min(buffer.readableBytes(), maxChunkSize);
        if (toRead > 0) {
            ByteBuf content = buffer.readRetainedSlice(toRead);
            out.add(new DefaultHttpContent(content));
        }
        return;
    }
    case READ_FIXED_LENGTH_CONTENT: {
        int readLimit = buffer.readableBytes();

        // Check if the buffer is readable first as we use the readable byte count
        // to create the HttpChunk. This is needed as otherwise we may end up with
        // create an HttpChunk instance that contains an empty buffer and so is
        // handled like it is the last HttpChunk.
        //
        // See https://github.com/netty/netty/issues/433
        if (readLimit == 0) {
            return;
        }

        int toRead = Math.min(readLimit, maxChunkSize);
        if (toRead > chunkSize) {
            toRead = (int) chunkSize;
        }
        ByteBuf content = buffer.readRetainedSlice(toRead);
        chunkSize -= toRead;

        if (chunkSize == 0) {
            // Read all content.
            out.add(new DefaultLastHttpContent(content, validateHeaders));
            resetNow();
        } else {
            out.add(new DefaultHttpContent(content));
        }
        return;
    }
    /**
     * everything else after this point takes care of reading chunked content. basically, read chunk size,
     * read chunk, read and ignore the CRLF and repeat until 0
     */
    case READ_CHUNK_SIZE: try {
        AppendableCharSequence line = lineParser.parse(buffer);
        if (line == null) {
            return;
        }
        int chunkSize = getChunkSize(line.toString());
        this.chunkSize = chunkSize;
        if (chunkSize == 0) {
            currentState = State.READ_CHUNK_FOOTER;
            return;
        }
        currentState = State.READ_CHUNKED_CONTENT;
        // fall-through
    } catch (Exception e) {
        out.add(invalidChunk(buffer, e));
        return;
    }
    case READ_CHUNKED_CONTENT: {
        assert chunkSize <= Integer.MAX_VALUE;
        int toRead = Math.min((int) chunkSize, maxChunkSize);
        if (!allowPartialChunks && buffer.readableBytes() < toRead) {
            return;
        }
        toRead = Math.min(toRead, buffer.readableBytes());
        if (toRead == 0) {
            return;
        }
        HttpContent chunk = new DefaultHttpContent(buffer.readRetainedSlice(toRead));
        chunkSize -= toRead;

        out.add(chunk);

        if (chunkSize != 0) {
            return;
        }
        currentState = State.READ_CHUNK_DELIMITER;
        // fall-through
    }
    case READ_CHUNK_DELIMITER: {
        final int wIdx = buffer.writerIndex();
        int rIdx = buffer.readerIndex();
        while (wIdx > rIdx) {
            byte next = buffer.getByte(rIdx++);
            if (next == HttpConstants.LF) {
                currentState = State.READ_CHUNK_SIZE;
                break;
            }
        }
        buffer.readerIndex(rIdx);
        return;
    }
    case READ_CHUNK_FOOTER: try {
        LastHttpContent trailer = readTrailingHeaders(buffer);
        if (trailer == null) {
            return;
        }
        out.add(trailer);
        resetNow();
        return;
    } catch (Exception e) {
        out.add(invalidChunk(buffer, e));
        return;
    }
    case BAD_MESSAGE: {
        // Keep discarding until disconnection.
        buffer.skipBytes(buffer.readableBytes());
        break;
    }
    case UPGRADED: {
        int readableBytes = buffer.readableBytes();
        if (readableBytes > 0) {
            // Keep on consuming as otherwise we may trigger an DecoderException,
            // other handler will replace this codec with the upgraded protocol codec to
            // take the traffic over at some point then.
            // See https://github.com/netty/netty/issues/2173
            out.add(buffer.readBytes(readableBytes));
        }
        break;
    }
    default:
        break;
    }
}

protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
    ByteBuf buf = null;
    if (msg instanceof HttpMessage) {
        if (state != ST_INIT) {
            throw new IllegalStateException("unexpected message type: " + StringUtil.simpleClassName(msg)
                    + ", state: " + state);
        }

        @SuppressWarnings({ "unchecked", "CastConflictsWithInstanceof" })
        H m = (H) msg;

        buf = ctx.alloc().buffer((int) headersEncodedSizeAccumulator);
        // Encode the message.
        encodeInitialLine(buf, m);
        state = isContentAlwaysEmpty(m) ? ST_CONTENT_ALWAYS_EMPTY :
                HttpUtil.isTransferEncodingChunked(m) ? ST_CONTENT_CHUNK : ST_CONTENT_NON_CHUNK;

        sanitizeHeadersBeforeEncode(m, state == ST_CONTENT_ALWAYS_EMPTY);

        encodeHeaders(m.headers(), buf);
        ByteBufUtil.writeShortBE(buf, CRLF_SHORT);

        headersEncodedSizeAccumulator = HEADERS_WEIGHT_NEW * padSizeForAccumulation(buf.readableBytes()) +
                                        HEADERS_WEIGHT_HISTORICAL * headersEncodedSizeAccumulator;
    }

    // Bypass the encoder in case of an empty buffer, so that the following idiom works:
    //
    //     ch.write(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    //
    // See https://github.com/netty/netty/issues/2983 for more information.
    if (msg instanceof ByteBuf) {
        final ByteBuf potentialEmptyBuf = (ByteBuf) msg;
        if (!potentialEmptyBuf.isReadable()) {
            out.add(potentialEmptyBuf.retain());
            return;
        }
    }

    if (msg instanceof HttpContent || msg instanceof ByteBuf || msg instanceof FileRegion) {
        switch (state) {
            case ST_INIT:
                throw new IllegalStateException("unexpected message type: " + StringUtil.simpleClassName(msg)
                    + ", state: " + state);
            case ST_CONTENT_NON_CHUNK:
                final long contentLength = contentLength(msg);
                if (contentLength > 0) {
                    if (buf != null && buf.writableBytes() >= contentLength && msg instanceof HttpContent) {
                        // merge into other buffer for performance reasons
                        buf.writeBytes(((HttpContent) msg).content());
                        out.add(buf);
                    } else {
                        if (buf != null) {
                            out.add(buf);
                        }
                        out.add(encodeAndRetain(msg));
                    }

                    if (msg instanceof LastHttpContent) {
                        state = ST_INIT;
                    }

                    break;
                }

                // fall-through!
            case ST_CONTENT_ALWAYS_EMPTY:

                if (buf != null) {
                    // We allocated a buffer so add it now.
                    out.add(buf);
                } else {
                    // Need to produce some output otherwise an
                    // IllegalStateException will be thrown as we did not write anything
                    // Its ok to just write an EMPTY_BUFFER as if there are reference count issues these will be
                    // propagated as the caller of the encode(...) method will release the original
                    // buffer.
                    // Writing an empty buffer will not actually write anything on the wire, so if there is a user
                    // error with msg it will not be visible externally
                    out.add(Unpooled.EMPTY_BUFFER);
                }

                break;
            case ST_CONTENT_CHUNK:
                if (buf != null) {
                    // We allocated a buffer so add it now.
                    out.add(buf);
                }
                encodeChunkedContent(ctx, msg, contentLength(msg), out);

                break;
            default:
                throw new Error();
        }

        if (msg instanceof LastHttpContent) {
            state = ST_INIT;
        }
    } else if (buf != null) {
        out.add(buf);
    }
}