HTTP POST example (REST) about rxnetty HOT 4 OPEN

import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.reactivex.netty.protocol.http.client.HttpClient;
import io.reactivex.netty.protocol.http.server.HttpServer;
import io.reactivex.netty.protocol.http.server.HttpServerRequest;
import io.reactivex.netty.protocol.http.server.HttpServerResponse;
import io.reactivex.netty.util.CollectBytes;
import org.pk11.rxnetty.router.Router;
import rx.Observable;

import java.nio.charset.Charset;

import static org.pk11.rxnetty.router.Dispatch.using;

public class PostExample {

  public static void main(String[] args) {

    HttpServer<ByteBuf, ByteBuf> server = HttpServer.newServer()
      .start(
        using(
          new Router<ByteBuf, ByteBuf>()
            .POST(
              "/echo",
              new EchoHandler()
            )
            .notFound(new Handler404())
        )
      );

    HttpClient.newClient("localhost", server.getServerPort())
      .createPost("/echo")
      .writeStringContent(
        Observable.range(1, 10).map(i -> "This is test " + i + "\n"),
        s -> true // flush on each selector, see below
      )
      .flatMap(
        response ->
          response.getContent().compose(CollectBytes.all())
            .map(
              b -> {
                String s = b.toString(Charset.defaultCharset());
                b.release();
                return s;
              }
            )
      )
      .subscribe(
        System.out::println,
        System.err::println
      );

    server.awaitShutdown();
  }

  private static class EchoHandler implements io.reactivex.netty.protocol.http.server.RequestHandler<ByteBuf, ByteBuf> {

    @Override
    public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
      // Three notes here:
      // 1) we have to release the content manually. Where you only need to know
      //    about the completion of processing, you can also use autoRelease()
      // 2) flushing is also manual - we are using a naive flush algorithm to
      //    prevent the response blocking once the write buffer is full.
      // 3) it is not guaranteed the ByteBuf block boundaries are the same as those sent
      return response.writeStringAndFlushOnEach(
        request.getContent()
          .map(
            b -> {
              String s = b.toString(Charset.defaultCharset());
              b.release();
              return s;
            }
          )
      );
    }
  }

  private static class Handler404 implements io.reactivex.netty.protocol.http.server.RequestHandler<ByteBuf, ByteBuf> {
    @Override
    public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
      System.out.println("Not found: " + request.getUri());
      return response.setStatus(HttpResponseStatus.NOT_FOUND)
        .writeString(Observable.just("Not found"));
    }
  }
}