Ring サーバーを実装する

どうも、僕です。
最近、Clojure 界隈から遠ざかっておりネタが全く思い浮かばなかったので
過去にプロトタイプでつくった Ring Server の話を書きたいと思います。
社用で使う用のプロトタイプなので全ては公開しませんが、気が向いたら公開するかも
知れません。

Ring SPEC

Ring は Clojure で Web アプリケーション 開発する際の規格のようなものです。
python の wsgi などの影響を受けており非常にシンプルな仕組みになっています。

ここでは詳細は割愛します。以下を参考にして下さい。

サーバーの実装の方針

割と使う側の話はあるのですがサーバー側実装についてはあまり情報ないかも知れません。
多くの場合は以下の2パターンになるでしょう。

• Servlet API から Ring へのマッピング
• 独自実装

Servlet API からのマッピングはあまり面白くなく、Servlet コンテナも必要になるので
少々面倒です。

uberjar で 1ファイルで配布してしまいたいので独自実装で実装していきます。
速度面を考え、コアは netty を使用し、その上に Ring 部を実装していきます。
呼び出し回数の少ない箇所や、動的に変更したい設定部分は Clojure で実装、コアは Java
で実装します。

Java -> Clojure

まず Java 側の実装です。

Clojure 連携の基本

Clojure から関数（Ring Handler）を受け取り、引数を Java で作成し実行します。
以下は Ring Handler を実行する例です

Map<Keyword, Object> response = (Map<Keyword, Object>) handler.invoke(PersistentArrayMap.create(reqMap));

Clojure の関数は IFn として表されます。
関数実行は invoke メソッドでできます。
引数部は Clojure 側で使用する型に合わせなければなりませんが、多くの場合は簡単に
変換できるようヘルパーメソッドが用意されています。

Ring Response は Clojure 側の map (PersistentArrayMap) で返ってくるわけですが、
Map インターフェイスで扱えば特に意識する必要はないでしょう。

Keyword

Ring のリクエストは キーが Keyword な map で表します。
Keyword は以下のように作成できます。

public static final Keyword HTTP = Keyword.intern("http");

また使用する Keyword はほぼ固定なので使いまわせるように static で作成しておく
方がよいでしょう。

public static final Keyword BODY_KEY = Keyword.intern("body");
public static final Keyword URI_KEY = Keyword.intern("uri");
public static final Keyword QUERY_KEY = Keyword.intern("query-string");
public static final Keyword REQUEST_METHOD_KEY = Keyword.intern("request-method");
public static final Keyword SERVER_NAME_KEY = Keyword.intern("server-name");

public static final Keyword SERVER_PORT_KEY = Keyword.intern("server-port");
public static final Keyword REMOTE_PORT_KEY = Keyword.intern("remote-port");
public static final Keyword SCHEME_KEY = Keyword.intern("scheme");
public static final Keyword CONTENT_TYPE_KEY = Keyword.intern("content-type");
public static final Keyword CONTENT_LENGTH_KEY = Keyword.intern("content-length");
public static final Keyword CHAR_ENC_KEY = Keyword.intern("charactor-encoding");
public static final Keyword HEADERS_KEY = Keyword.intern("headers");

public static final Keyword HTTP = Keyword.intern("http");
public static final Keyword STATUS = Keyword.intern("status");
public static final Keyword USE_SEND_FILE = Keyword.intern("use-send-file");
static {
    METHOD_MAP = new HashMap<>();
    METHOD_MAP.put(HttpMethod.GET, Keyword.intern("get"));
    METHOD_MAP.put(HttpMethod.POST, Keyword.intern("post"));
    METHOD_MAP.put(HttpMethod.PUT, Keyword.intern("put"));
    METHOD_MAP.put(HttpMethod.PATCH, Keyword.intern("patch"));
    METHOD_MAP.put(HttpMethod.OPTIONS, Keyword.intern("options"));
    METHOD_MAP.put(HttpMethod.DELETE, Keyword.intern("delete"));
    METHOD_MAP.put(HttpMethod.HEAD, Keyword.intern("head"));
    METHOD_MAP.put(HttpMethod.CONNECT, Keyword.intern("connect"));
    METHOD_MAP.put(HttpMethod.TRACE, Keyword.intern("trace"));
}

Ring Request

Ring Request は上記でもあるように PersistentArrayMap です。
PersistentArrayMap は Map から作成できるため、Map で作成後、最後に変換します。

private Map<Keyword, Object> createRingRequest(final ChannelHandlerContext ctx, final FullHttpRequest request) {

    Map<Keyword, Object> m = new HashMap<>();
    // :scheme
    m.put(SCHEME_KEY, HTTP);
    // :uri
    // :query
    setURIAndQuery(m, request);
    // :server-name
    // :server-port
    // :content-type
    // :content-length
    // :charctor-encoding
    setServerNames(m, request);
    // :request-method
    m.put(REQUEST_METHOD_KEY, getHttpMethod(request));
    // :remote-address
    m.put(REMOTE_PORT_KEY, getRemoteAddress(ctx, request));
    // :headers
    m.put(HEADERS_KEY, getRequestHeaders(request));

    return m;
}

Ring Response

Ring Response は複数の型をサポートしています。
そのため型を調べ、実装します。

private void writeRingResponse(Channel ch, Map<Keyword, Object> ringResponse) throws Exception {

    Integer statusCode = ((Long) ringResponse.get(STATUS)).intValue();
    if (statusCode == null) {
        statusCode = 200;
    }
    HttpResponseStatus status = HttpResponseStatus.valueOf(statusCode);
    FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status);

    setResponseHeaders(response, ringResponse);

    Object body = ringResponse.get(BODY_KEY);

    if (body != null) {
        sendResponse(body, response, ch, keepAlive);
    } else {
        throw new IllegalArgumentException("response body is null");
    }
}

private void setResponseHeaders(HttpResponse response, Map<Keyword, Object> ringResponse) {
    HttpHeaders headers = response.headers();

    @SuppressWarnings("unchecked")
        final Map<String, Object> headerMap = (Map<String, Object>) ringResponse.get(HEADERS_KEY);

    if (headerMap == null) {
        return;
    }

    for (Map.Entry<String, Object> header : headerMap.entrySet()) {
        headers.set(header.getKey(), (String)header.getValue());
    }

    if (keepAlive) {
        headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
    }

}

private void sendResponse(Object body, FullHttpResponse response, Channel ch, boolean keepAlive) throws Exception {

    // System.out.println(Thread.currentThread());
    if (body instanceof String) {
        sendStringResponse((String) body, response, ch, keepAlive);
    } else if (body instanceof ISeq) {
        StringBuilder sb = new StringBuilder();
        for (ISeq s = (ISeq) body; s != null; s = s.next()) {
            String str = (String) s.first();
            sb.append(str);
        }
        sendStringResponse(sb.toString(), response, ch, keepAlive);
    } else if (body instanceof InputStream) {
        sendInputStreamResponse((InputStream) body, response, ch, keepAlive);
    } else if (body instanceof File) {
        sendFileStreamResponse((File) body, response, ch, keepAlive);
    } else {
        //TODO WARN or 404 ?
        sendStringResponse("", response, ch, keepAlive);
    }
}

Clojure -> Java

Clojure 側は設定、組み立て部です。
基本はラッパーを作るだけです。
組み立て部は Clojure で書いた方が楽でしょう。

ラッパー

各種ラッパーを定義します。
Clojure から短い名前で呼び出せるようにしています。

(ns arianrod.netty
  (:import
   [java.nio.charset Charset]
   [io.netty.bootstrap ServerBootstrap Bootstrap]
   [io.netty.channel EventLoop EventLoopGroup ChannelHandler ChannelPipeline ChannelHandlerContext]
   [arianrod.support CljChannelInHandler BootstrapFactory ChannelHandlerContextUtil CljSimpleChannelHandler UnpooledWrapper
    ChunkedInputStream ChannelEventHandler]))

(defn- is-linux []
  (= "Linux" (System/getProperty "os.name")))

(defmacro ^EventLoopGroup create-elg []
  (if (is-linux)
    `(io.netty.channel.epoll.EpollEventLoopGroup.)
    `(io.netty.channel.nio.NioEventLoopGroup.)))

(defmacro ^EventLoop create-el []
  (if (is-linux)
    `(io.netty.channel.epoll.EpollEventLoop.)
    `(io.netty.channel.nio.NioEventLoop.)))

(defmacro ^EventLoop create-nssc []
  (if (is-linux)
    `io.netty.channel.epoll.EpollServerSocketChannel
    `io.netty.channel.socket.nio.NioServerSocketChannel))

(defmacro ^EventLoop create-nsc []
  (if (is-linux)
    `io.netty.channel.epoll.EpollSocketChannel
    `io.netty.channel.socket.nio.NioSocketChannel))

(defmacro ^ChannelHandler channel-in-handler [m]
  `(CljChannelInHandler. ~m))

(defn ^ChannelHandler make-handler [fn & {:keys [filter] :or {filter [:read :complete :exception]}}]
  (ChannelEventHandler. fn (set filter)))

(defmacro ^ChannelHandler simple-handler [m]
  `(CljSimpleChannelHandler. ~m))

(defmacro write-ctx [ctx msg]
  `(ChannelHandlerContextUtil/writeCtx ~ctx ~msg))

(defmacro flush-ctx [ctx]
  `(ChannelHandlerContextUtil/flushCtx ~ctx))

(defmacro write-flush-ctx [ctx msg]
  `(ChannelHandlerContextUtil/writeAndFlushCtx ~ctx ~msg))

(defmacro close-ctx [ctx]
  `(ChannelHandlerContextUtil/closeCtx ~ctx))

(defn ^ServerBootstrap server-factory [fn & {:keys [filter] :or {filter [:read :complete :exception]}}]
  (BootstrapFactory/createSimpleServerBootstrap (create-nssc) fn (set filter)))

(defn ^ServerBootstrap server-factory-init [initializer]
  (BootstrapFactory/createServerBootstrap (create-nssc) initializer))

(defn ^Bootstrap client-factory [initializer]
  (BootstrapFactory/createBootstrap (create-nsc) initializer))

(defmacro add-last-pipeline [pipline name fn]
  `(BootstrapFactory/addLastPipeline ~pipline ~name ~fn))

;; Buffer API

(defn wrap-buffer-bytes [a]
  (UnpooledWrapper/wrappedBufferBytes a))

(defn copied-buffer [^String string ^Charset charset]
  (UnpooledWrapper/copiedBuffer string charset))

(defn create-buf [cap]
  (UnpooledWrapper/createBuffer cap))

(defn create-direct-buf [cap]
  (UnpooledWrapper/createDirectBuffer cap))

(defn ->byte-array [s]
  (byte-array (map (comp byte int) s)))

(defn ->bytes [s]
  (bytes (byte-array (map (comp byte int) s))))

内部的に EventLoop の種類などを自動で切り替えます。

Server

起動部分はほぼ netty のお作法通りです。

(ns arianrod.server
  (:import
   [io.netty.bootstrap ServerBootstrap]
   [io.netty.channel EventLoopGroup ChannelFuture])
  (:require
   [arianrod.netty :as netty]))

(def ^:dynamic *debug* false)
(def ^:dynamic *backlog* 1024)

(defn wait-future [^ChannelFuture future ^EventLoopGroup boss ^EventLoopGroup work]
  (let [^ChannelFuture cf (-> (.channel future)
                              (.closeFuture))]
    (fn []
      (try
        (.sync cf)
        (finally
          (.shutdownGracefully boss)
          (.shutdownGracefully work))))))

(defn ^ServerBootstrap bind-server [^ServerBootstrap bootstrap ^long port]
  (-> (.bind bootstrap port)
      (.sync)))

(defn start-server* [factory port & [boss work]]
  (let [boss (or boss (netty/create-elg))
        work (or work (netty/create-elg))]
    (-> (factory)
        (.group boss work)
        (bind-server port)
        (wait-future boss work))))

(defn start-server [factory port & {:keys [wait] :or {wait true}}]
  (let [waiter (start-server* factory port)]
    (if wait
      (waiter)
      waiter)))

(defmacro start-server-wait [factory port]
  `(start-server ~factory ~port :wait true))

(ns arianrod.ring
  (:use arianrod.netty)
  (:import
   [io.netty.bootstrap ServerBootstrap]
   [arianrod.support.ring RingBootstrapFactory RingChannelHandler]))

(def ^:dynamic *option* {:debug false
                         :backlog 1024
                         :use-send-file true})

(defn ring-server-factory [handler options]
  (RingBootstrapFactory/createRingServerBootstrap
   (create-nssc) options handler))

(defn create-ring-server
  ([handler] (create-ring-server handler *option*))
  ([handler options]
     (ring-server-factory handler (merge *option* options))))

(defn create-http-server []
  (create-ring-server
   (fn [req]
     {:status 200
      :headers {"Content-Type" "text/plain"}
      :body "hello world"})))

全体

package arianrod.support.ring;

import clojure.lang.*;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.stream.ChunkedFile;
import io.netty.handler.stream.ChunkedStream;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.HashMap;
import java.util.Map;

public class RingChannelHandler extends SimpleChannelInboundHandler<HttpObject> {

    public static final Keyword BODY_KEY = Keyword.intern("body");
    public static final Keyword URI_KEY = Keyword.intern("uri");
    public static final Keyword QUERY_KEY = Keyword.intern("query-string");
    public static final Keyword REQUEST_METHOD_KEY = Keyword.intern("request-method");
    public static final Keyword SERVER_NAME_KEY = Keyword.intern("server-name");

    public static final Keyword SERVER_PORT_KEY = Keyword.intern("server-port");
    public static final Keyword REMOTE_PORT_KEY = Keyword.intern("remote-port");
    public static final Keyword SCHEME_KEY = Keyword.intern("scheme");
    public static final Keyword CONTENT_TYPE_KEY = Keyword.intern("content-type");
    public static final Keyword CONTENT_LENGTH_KEY = Keyword.intern("content-length");
    public static final Keyword CHAR_ENC_KEY = Keyword.intern("charactor-encoding");
    public static final Keyword HEADERS_KEY = Keyword.intern("headers");

    public static final Keyword HTTP = Keyword.intern("http");
    public static final Keyword STATUS = Keyword.intern("status");
    public static final Keyword USE_SEND_FILE = Keyword.intern("use-send-file");

    private static final String CHARSET = "charset=";
    private static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");

    private static final int BODY_MEMORY_LIMIT = 1024 * 1024 * 16;

    private static final Map<HttpMethod, Keyword> METHOD_MAP;

    static {
        METHOD_MAP = new HashMap<>();
        METHOD_MAP.put(HttpMethod.GET, Keyword.intern("get"));
        METHOD_MAP.put(HttpMethod.POST, Keyword.intern("post"));
        METHOD_MAP.put(HttpMethod.PUT, Keyword.intern("put"));
        METHOD_MAP.put(HttpMethod.PATCH, Keyword.intern("patch"));
        METHOD_MAP.put(HttpMethod.OPTIONS, Keyword.intern("options"));
        METHOD_MAP.put(HttpMethod.DELETE, Keyword.intern("delete"));
        METHOD_MAP.put(HttpMethod.HEAD, Keyword.intern("head"));
        METHOD_MAP.put(HttpMethod.CONNECT, Keyword.intern("connect"));
        METHOD_MAP.put(HttpMethod.TRACE, Keyword.intern("trace"));
    }

    private final APersistentMap option;
    private final IFn handler;
    private Map<Keyword, Object> reqMap;
    private ByteBuf byteBuf = null;
    private Path tempFilePath = null;
    private boolean keepAlive;

    public RingChannelHandler(final IFn handler, final APersistentMap option) {
        this.handler = handler;
        this.option = option;
    }

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

    private void callRingHandler(ChannelHandlerContext ctx, InputStream input) throws Exception {
        //System.out.println(Thread.currentThread());

        try (InputStream in = input) {
                reqMap.put(BODY_KEY, in);

                @SuppressWarnings("unchecked")
                Map<Keyword, Object> response = (Map<Keyword, Object>) handler.invoke(PersistentArrayMap.create(reqMap));
                if (response != null) {
                    writeRingResponse(ctx.channel(), response);
                } else {
                    throw new IllegalArgumentException("response is null");
                }

            }
    }

    @Override
        public void messageReceived(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
        if (msg instanceof HttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;
            keepAlive = HttpHeaderUtil.isKeepAlive(request);

            Long contentLength = HttpHeaderUtil.getContentLength(request, 0);
            HttpMethod method = request.method();
            reqMap = this.createRingRequest(ctx, request);

            if (method == HttpMethod.POST || method == HttpMethod.PUT) {
                if (contentLength > BODY_MEMORY_LIMIT) {
                    // large data
                    tempFilePath = Files.createTempFile(null, null);

                } else {
                    // memory
                    byteBuf = Unpooled.directBuffer(contentLength.intValue());
                }
            } else {
                callRingHandler(ctx, new ByteBufInputStream(request.content()));
            }
        }

        if (msg instanceof HttpContent) {
            HttpContent content = (HttpContent) msg;
            if (byteBuf != null) {
                byteBuf.writeBytes(content.content());

                if (msg instanceof LastHttpContent) {
                    callRingHandler(ctx, new ByteBufInputStream(byteBuf));
                }
            } else if (tempFilePath != null) {
                Files.write(tempFilePath, content.content().array(), StandardOpenOption.APPEND);

                if (msg instanceof LastHttpContent) {
                    callRingHandler(ctx, Files.newInputStream(tempFilePath, StandardOpenOption.DELETE_ON_CLOSE));
                }
            }
        }
    }

    private Keyword getHttpMethod(final FullHttpRequest request) {
        HttpMethod method = request.method();
        if (METHOD_MAP.containsKey(method)) {
            return METHOD_MAP.get(method);
        }
        return Keyword.intern(null, String.valueOf(method.name().toLowerCase()));
    }

    private void setURIAndQuery(final Map<Keyword, Object> m, final FullHttpRequest request) {
        String url = request.uri();
        int idx = url.indexOf('?');
        String uri;
        String queryString;
        if (idx > 0) {
            uri = url.substring(0, idx);
            queryString = url.substring(idx + 1);
        } else {
            uri = url;
            queryString = null;
        }
        m.put(URI_KEY, uri);
        m.put(QUERY_KEY, queryString);
    }

    private void setServerNames(final Map<Keyword, Object> m, final FullHttpRequest request) {
        HttpHeaders headers = request.headers();
        String serverName = null;
        Integer serverPort = null;
        String contentType = null;
        String charset = null;

        String h = (String) headers.get("host");
        if (h != null) {
            int idx = h.lastIndexOf(':');
            if (idx != -1) {
                serverName = h.substring(0, idx);
                serverPort = Integer.valueOf(h.substring(idx + 1));
            } else {
                serverName = h;
            }
        }

        String ct = (String) headers.get(HttpHeaderNames.CONTENT_TYPE);
        if (ct != null) {
            int idx = ct.indexOf(";");
            if (idx != -1) {
                int cidx = ct.indexOf(CHARSET, idx);
                if (cidx != -1) {
                    contentType = ct.substring(0, idx);
                    charset = ct.substring(cidx + CHARSET.length());
                } else {
                    contentType = ct;
                }
            } else {
                contentType = ct;
            }
        }

        m.put(SERVER_NAME_KEY, serverName);
        m.put(SERVER_PORT_KEY, serverPort);
        m.put(CONTENT_TYPE_KEY, contentType);

        long length = HttpHeaderUtil.getContentLength(request);
        if (length > 0) {
            m.put(CONTENT_LENGTH_KEY, length);
        } else {
            m.put(CONTENT_LENGTH_KEY, null);
        }
        m.put(CHAR_ENC_KEY, charset);
    }

    private String getRemoteAddress(final ChannelHandlerContext ctx, final FullHttpRequest request) {
        String h = (String) request.headers().get("X-Forward-For");
        if (h != null) {
            int idx = h.indexOf(',');
            if (idx == -1) {
                return h;
            } else {
                // X-Forwarded-For: client, proxy1, proxy2
                return h.substring(0, idx);
            }
        }
        InetSocketAddress sockAddr = (InetSocketAddress) ctx.channel().remoteAddress();
        return sockAddr.getAddress().getHostAddress();
    }

    private IPersistentMap getRequestHeaders(final FullHttpRequest request) {
        HttpHeaders headers = request.headers();
        Map<String, String> m = new HashMap<>();
        for (Map.Entry<CharSequence, CharSequence> header : headers.entries()) {
            String lo = header.getKey().toString().toLowerCase();
            m.put(lo, (String)header.getValue());
        }
        return PersistentArrayMap.create(m);
    }

    private Map<Keyword, Object> createRingRequest(final ChannelHandlerContext ctx, final FullHttpRequest request) {

        Map<Keyword, Object> m = new HashMap<>();
        // :scheme
        m.put(SCHEME_KEY, HTTP);
        // :uri
        // :query
        setURIAndQuery(m, request);
        // :server-name
        // :server-port
        // :content-type
        // :content-length
        // :charctor-encoding
        setServerNames(m, request);
        // :request-method
        m.put(REQUEST_METHOD_KEY, getHttpMethod(request));
        // :remote-address
        m.put(REMOTE_PORT_KEY, getRemoteAddress(ctx, request));
        // :headers
        m.put(HEADERS_KEY, getRequestHeaders(request));

        return m;
    }

    private void writeRingResponse(Channel ch, Map<Keyword, Object> ringResponse) throws Exception {

        Integer statusCode = ((Long) ringResponse.get(STATUS)).intValue();
        if (statusCode == null) {
            statusCode = 200;
        }
        HttpResponseStatus status = HttpResponseStatus.valueOf(statusCode);
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status);

        setResponseHeaders(response, ringResponse);

        Object body = ringResponse.get(BODY_KEY);

        if (body != null) {
            sendResponse(body, response, ch, keepAlive);
        } else {
            throw new IllegalArgumentException("response body is null");
        }
    }

    private void setResponseHeaders(HttpResponse response, Map<Keyword, Object> ringResponse) {
        HttpHeaders headers = response.headers();

        @SuppressWarnings("unchecked")
            final Map<String, Object> headerMap = (Map<String, Object>) ringResponse.get(HEADERS_KEY);

        if (headerMap == null) {
            return;
        }

        for (Map.Entry<String, Object> header : headerMap.entrySet()) {
            headers.set(header.getKey(), (String)header.getValue());
        }

        if (keepAlive) {
            headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        }

    }

    private void sendResponse(Object body, FullHttpResponse response, Channel ch, boolean keepAlive) throws Exception {

        // System.out.println(Thread.currentThread());
        if (body instanceof String) {
            sendStringResponse((String) body, response, ch, keepAlive);
        } else if (body instanceof ISeq) {
            StringBuilder sb = new StringBuilder();
            for (ISeq s = (ISeq) body; s != null; s = s.next()) {
                String str = (String) s.first();
                sb.append(str);
            }
            sendStringResponse(sb.toString(), response, ch, keepAlive);
        } else if (body instanceof InputStream) {
            sendInputStreamResponse((InputStream) body, response, ch, keepAlive);
        } else if (body instanceof File) {
            sendFileStreamResponse((File) body, response, ch, keepAlive);
        } else {
            //TODO WARN or 404 ?
            sendStringResponse("", response, ch, keepAlive);
        }
    }

    private void setContentLength(FullHttpResponse response, long length) {
        HttpHeaders headers = response.headers();
        CharSequence val = headers.get(HttpHeaderNames.CONTENT_LENGTH);
        if (val == null || val.equals("")) {
            headers.setLong(HttpHeaderNames.CONTENT_LENGTH, length);
        }
    }

    private Charset getResponseCharset(FullHttpResponse response) {
        HttpHeaders headers = response.headers();
        String type = (String)headers.get(HttpHeaderNames.CONTENT_TYPE);
        if (type != null) {
            try {
                type = type.toLowerCase();
                int i = type.indexOf(CHARSET);
                if (i != -1) {
                    String charset = type.substring(i + CHARSET.length()).trim();
                    return Charset.forName(charset);
                }
            } catch (Exception ignore) {
            }
        }
        return DEFAULT_CHARSET;
    }

    private void sendStringResponse(final String body, final FullHttpResponse response, Channel ch, boolean keepAlive) {
        Charset charset = getResponseCharset(response);
        ByteBuf buffer = Unpooled.copiedBuffer(body, charset);
        setContentLength(response, buffer.readableBytes());
        response.content().writeBytes(buffer);
        buffer.release();
        if (keepAlive) {
            ch.write(response);
            //ctx.write(response).addListener(ChannelFutureListener.CLOSE);
        } else {
            ch.write(response).addListener(ChannelFutureListener.CLOSE);
        }
    }

    private void sendInputStreamResponse(final InputStream stream, final FullHttpResponse response, Channel ch, boolean keepAlive) {
        ch.write(response);
        ChannelFuture future = ch.write(new ChunkedStream(stream));

        future.addListener(new GenericFutureListener<Future<Void>>() {
                @Override
                    public void operationComplete(Future<Void> f) throws Exception {
                    stream.close();
                }
            });
        future.addListener(ChannelFutureListener.CLOSE);
    }

    private void sendFileStreamResponse(final File file, final FullHttpResponse response, Channel ch, boolean keepAlive) throws Exception {

        Boolean useSendFile = (Boolean) option.get(USE_SEND_FILE);

        RandomAccessFile raf;
        try {
            raf = new RandomAccessFile(file, "r");
        } catch (FileNotFoundException e) {
            return;
        }
        long fileLength = raf.length();
        setContentLength(response, fileLength);
        ch.write(response);

        // Write the content.
        if (useSendFile != null && useSendFile) {
            ch.write(new DefaultFileRegion(raf.getChannel(), 0, fileLength), ch.newProgressivePromise());
        } else {
            ch.write(new ChunkedFile(raf, 0, fileLength, 8192), ch.newProgressivePromise());
        }

        ChannelFuture lastContentFuture = ch.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
        if (!keepAlive) {
            lastContentFuture.addListener(ChannelFutureListener.CLOSE);
        }

    }

    static final class RingHandler implements Runnable {

        private final RingChannelHandler ringChannelHandler;
        private final Channel ch;
        private final InputStream in;

        public RingHandler(Channel ch, RingChannelHandler handler, InputStream in) {
            this.ringChannelHandler = handler;
            this.ch = ch;
            this.in = in;
        }

        @Override
        public void run() {

            try (InputStream input = this.in) {
                    this.ringChannelHandler.reqMap.put(BODY_KEY, input);
                    @SuppressWarnings("unchecked")
                        Map<Keyword, Object> response = (Map<Keyword, Object>) this.ringChannelHandler.handler.invoke(PersistentArrayMap.create(this.ringChannelHandler.reqMap));
                    if (response != null) {
                        this.ringChannelHandler.writeRingResponse(this.ch, response);
                    } else {
                        throw new IllegalArgumentException("response is null");
                    }
            } catch (Exception e) {
                //TODO Logging
                e.printStackTrace();
            } finally {
                this.ch.close();
            }

        }
    }

}

package arianrod.support.ring;

import clojure.lang.APersistentMap;
import clojure.lang.IFn;
import clojure.lang.Keyword;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ServerChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class RingBootstrapFactory {

    public static final Keyword DEBUG_KEY = Keyword.intern("debug");
    public static final Keyword BACKLOG_KEY = Keyword.intern("backlog");

    public static ServerBootstrap createRingServerBootstrap(Class<? extends ServerChannel> channelClass, final APersistentMap option, final IFn handler) {

        final Boolean debug = (Boolean) option.get(DEBUG_KEY);
        final int backlog = ((Long) option.get(BACKLOG_KEY)).intValue();

        ServerBootstrap b = new ServerBootstrap();
        b.channel(channelClass)
                .option(ChannelOption.SO_BACKLOG, backlog)
                .option(ChannelOption.TCP_NODELAY, true)
                .childHandler(new RingChannelInitializer(handler, option));

        if (debug) {
            b.handler(new LoggingHandler(LogLevel.DEBUG));
        }
        return b;
    }

    static class RingChannelInitializer extends ChannelInitializer<SocketChannel> {

        private final IFn handler;
        private final APersistentMap option;

        public RingChannelInitializer(final IFn handler, final APersistentMap option) {
            super();
            this.option = option;
            this.handler = handler;
        }

        @Override
        public void initChannel(final SocketChannel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new HttpRequestDecoder())
                    .addLast(new HttpObjectAggregator(65536))
                    .addLast(new HttpResponseEncoder())
                    .addLast(new ChunkedWriteHandler())
                    .addLast(new RingChannelHandler(handler, option));
        }
    }

}

package arianrod.support;

import clojure.lang.IFn;
import clojure.lang.Keyword;
import clojure.lang.PersistentArrayMap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.util.logging.Level;
import java.util.logging.Logger;

public class CljSimpleChannelHandler extends SimpleChannelInboundHandler {

    private static final Logger logger = Logger.getLogger(
                                                          CljSimpleChannelHandler.class.getName());

    private static final Keyword isSharableKey = Keyword.intern(null, "is-sharable");
    private static final Keyword channelReadCompKey = Keyword.intern(null, "channel-read-complete");
    private static final Keyword channelActiveKey = Keyword.intern(null, "channel-active");
    private static final Keyword messageReceivedKey = Keyword.intern(null, "message-received");
    private static final Keyword channelInactiveKey = Keyword.intern(null, "channel-in-active");
    private static final Keyword exceptionCaughtKey = Keyword.intern(null, "exception-caught");

    private final PersistentArrayMap map;

    public CljSimpleChannelHandler(PersistentArrayMap map) {
        this.map = map;
    }

    @Override
    public boolean isSharable() {
        if (this.map.containsKey(isSharableKey)) {
            final IFn f = (IFn) this.map.valAt(isSharableKey);
            return (Boolean) f.invoke();
        } else {
            return super.isSharable();
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        if (this.map.containsKey(channelActiveKey)) {
            final IFn f = (IFn) this.map.valAt(channelActiveKey);
            f.invoke(ctx);
        } else {
            super.channelActive(ctx);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (this.map.containsKey(channelInactiveKey)) {
            final IFn f = (IFn) this.map.valAt(channelInactiveKey);
            f.invoke(ctx);
        } else {
            super.channelInactive(ctx);
        }
    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (this.map.containsKey(messageReceivedKey)) {
            final IFn f = (IFn) this.map.valAt(messageReceivedKey);
            f.invoke(ctx, msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (this.map.containsKey(channelReadCompKey)) {
            final IFn f = (IFn) this.map.valAt(channelReadCompKey);
            f.invoke(ctx);
        } else {
            super.channelReadComplete(ctx);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (this.map.containsKey(exceptionCaughtKey)) {
            final IFn f = (IFn) this.map.valAt(exceptionCaughtKey);
            f.invoke(ctx, cause);
        } else {

            logger.log(Level.WARNING, "Unexpected exception from downstream.", cause);
            ctx.close();
        }
    }

}

package arianrod.support;

import clojure.lang.IFn;
import clojure.lang.Keyword;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

import java.util.Set;

public class ChannelEventHandler extends ChannelHandlerAdapter {

    public static final Keyword INACTIVE_KEY = Keyword.intern("inactive");
    public static final Keyword ACTIVE_KEY = Keyword.intern("active");
    public static final Keyword READ_KEY = Keyword.intern("read");
    public static final Keyword COMPLETE_KEY = Keyword.intern("complete");
    public static final Keyword EXCEPTION_KEY = Keyword.intern("exception");

    private EventSource eventSource;

    private final IFn handler;
    private final Set filter;

    public ChannelEventHandler(final IFn handler, final Set filter) {
        this.handler = handler;
        this.filter = filter;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (!this.filter.contains(INACTIVE_KEY)) {
            return;
        }
        if (eventSource == null) {
            this.eventSource = new EventSource(ctx, INACTIVE_KEY);
        } else {
            this.eventSource.ctx = ctx;
            this.eventSource.type = INACTIVE_KEY;
            this.eventSource.object = null;
        }
        this.handler.invoke(this.eventSource);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        if (!this.filter.contains(ACTIVE_KEY)) {
            return;
        }
        if (eventSource == null) {
            this.eventSource = new EventSource(ctx, ACTIVE_KEY);
        } else {
            this.eventSource.ctx = ctx;
            this.eventSource.type = ACTIVE_KEY;
            this.eventSource.object = null;
        }
        this.handler.invoke(this.eventSource);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (!this.filter.contains(COMPLETE_KEY)) {
            return;
        }
        if (eventSource == null) {
            this.eventSource = new EventSource(ctx, COMPLETE_KEY);

        } else {
            this.eventSource.ctx = ctx;
            this.eventSource.type = COMPLETE_KEY;
            this.eventSource.object = null;
        }
        this.handler.invoke(this.eventSource);


    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (!this.filter.contains(EXCEPTION_KEY)) {
            return;
        }
        if (eventSource == null) {
            this.eventSource = new EventSource(ctx, EXCEPTION_KEY, cause);

        } else {
            this.eventSource.ctx = ctx;
            this.eventSource.type = EXCEPTION_KEY;
            this.eventSource.object = cause;
        }
        this.handler.invoke(this.eventSource);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (!this.filter.contains(COMPLETE_KEY)) {
            return;
        }

        if (eventSource == null) {
            this.eventSource = new EventSource(ctx, READ_KEY, msg);

        } else {
            this.eventSource.ctx = ctx;
            this.eventSource.type = READ_KEY;
            this.eventSource.object = msg;
        }
        this.handler.invoke(this.eventSource);
    }

}

package arianrod.support;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;

public class ChannelHandlerContextUtil {

    public static final ChannelFuture writeCtx(ChannelHandlerContext ctx, Object msg) {
        return ctx.write(msg);
    }

    public static final ChannelHandlerContext flushCtx(ChannelHandlerContext ctx) {
        return ctx.flush();
    }

    public static final ChannelFuture closeCtx(ChannelHandlerContext ctx) {
        return ctx.close();
    }

    public static final ChannelFuture writeAndFlushCtx(ChannelHandlerContext ctx, Object msg) {
        return ctx.writeAndFlush(msg);
    }

}

package arianrod.support;

import clojure.lang.IFn;
import clojure.lang.Keyword;
import clojure.lang.PersistentArrayMap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelHandlerAdapter;

import java.util.logging.Level;
import java.util.logging.Logger;

public class CljChannelInHandler extends ChannelHandlerAdapter {

    private static final Logger logger = Logger.getLogger(
                                                          CljChannelInHandler.class.getName());

    private static final Keyword isSharableKey = Keyword.intern(null, "is-sharable");
    private static final Keyword channelReadKey = Keyword.intern(null, "channel-read");
    private static final Keyword channelReadCompKey = Keyword.intern(null, "channel-read-complete");
    private static final Keyword channelActiveKey = Keyword.intern(null, "channel-active");
    private static final Keyword channelInactiveKey = Keyword.intern(null, "channel-in-active");
    private static final Keyword exceptionCaughtKey = Keyword.intern(null, "exception-caught");

    private final PersistentArrayMap map;

    public CljChannelInHandler(PersistentArrayMap map) {
        this.map = map;
    }

    @Override
    public boolean isSharable() {
        if (this.map.containsKey(isSharableKey)) {
            final IFn f = (IFn) this.map.valAt(isSharableKey);
            return (Boolean) f.invoke();
        } else {
            return super.isSharable();
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        if (this.map.containsKey(channelActiveKey)) {
            final IFn f = (IFn) this.map.valAt(channelActiveKey);
            f.invoke(ctx);
        } else {
            super.channelActive(ctx);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (this.map.containsKey(channelInactiveKey)) {
            final IFn f = (IFn) this.map.valAt(channelInactiveKey);
            f.invoke(ctx);
        } else {
            super.channelInactive(ctx);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (this.map.containsKey(channelReadKey)) {
            final IFn f = (IFn) this.map.valAt(channelReadKey);
            f.invoke(ctx, msg);
        } else {
            super.channelRead(ctx, msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (this.map.containsKey(channelReadCompKey)) {
            final IFn f = (IFn) this.map.valAt(channelReadCompKey);
            f.invoke(ctx);
        } else {
            super.channelReadComplete(ctx);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (this.map.containsKey(exceptionCaughtKey)) {
            final IFn f = (IFn) this.map.valAt(exceptionCaughtKey);
            f.invoke(ctx, cause);
        } else {

            logger.log(Level.WARNING, "Unexpected exception from downstream.", cause);
            ctx.close();
        }
    }
}

package arianrod.support;

import clojure.lang.IDeref;
import clojure.lang.IFn;
import clojure.lang.Keyword;
import clojure.lang.RT;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.ChannelMatcher;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class EventSource implements IDeref {

    public static Map<String, DefaultChannelGroup> groupMap = new ConcurrentHashMap<>();

    public Keyword type;
    public Object object;
    public ChannelHandlerContext ctx;


    public EventSource(ChannelHandlerContext ctx, Keyword type) {
        this.ctx = ctx;
        this.type = type;
    }

    public EventSource(ChannelHandlerContext ctx, Keyword type, Object object) {
        this(ctx, type);
        this.object = object;
    }

    public void addGroup(String ns) {
        if (!groupMap.containsKey(ns)) {
            DefaultChannelGroup defaultChannelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
            groupMap.put(ns, defaultChannelGroup);
        }
    }

    public void joinGroup(String ns) {
        this.addGroup(ns);
        groupMap.get(ns).add(ctx.channel());
    }

    public ChannelGroupFuture putGroup(String ns, Object msg) {
        if (groupMap.containsKey(ns)) {
            return groupMap.get(ns).writeAndFlush(msg);
        }
        return null;
    }

    public ChannelGroupFuture putGroup(String ns, Object msg, final IFn matcher) {
        if (groupMap.containsKey(ns)) {
            return groupMap.get(ns).writeAndFlush(msg, new ChannelMatcher() {
                @Override
                public boolean matches(Channel channel) {
                    Object res = matcher.invoke(channel);
                    return res instanceof Boolean && (Boolean) res;
                }
            });
        }
        return null;
    }

    public ChannelGroupFuture closeGroup(String ns) {
        if (groupMap.containsKey(ns)) {
            return groupMap.get(ns).close();
        }
        return null;
    }

    public ChannelGroupFuture closeGroup(String ns, final IFn matcher) {
        if (groupMap.containsKey(ns)) {
            return groupMap.get(ns).close(new ChannelMatcher() {
                @Override
                public boolean matches(Channel channel) {
                    Object res = matcher.invoke(channel);
                    return res instanceof Boolean && (Boolean) res;
                }
            });
        }
        return null;
    }

    public void sendRemote(String host, int port, final IFn initializer, final IFn channelWriter) throws InterruptedException{
        Bootstrap b = BootstrapFactory.createBootstrap(this.ctx.channel().getClass(), initializer);
        b = b.group(this.ctx.channel().eventLoop());
        Channel channel = b.connect(host, port).sync().channel();
        channelWriter.invoke(channel);
        channel.closeFuture().sync();

    }

    @Override
    public Object deref() {
        return RT.vector(this.type, this.object);
    }

    public Object get() {
        return this.object;
    }

    public EventSource put(Object o) {
        this.ctx.write(o);
        return this;
    }

    public EventSource putFlush(Object o) {
        this.ctx.writeAndFlush(o);
        return this;
    }

    public EventSource flush() {
        this.ctx.flush();
        return this;
    }

}

package arianrod.support;

import clojure.lang.IFn;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;

public class SimpleChannelInitializer extends ChannelInitializer<SocketChannel> {

    private final IFn callback;

    public SimpleChannelInitializer(IFn callback) {
        this.callback = callback;
    }

    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        this.callback.invoke(ch.pipeline());
    }
}

package arianrod.support;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;

public class UnpooledWrapper {

    public static final ByteBuf wrappedBufferBytes(byte[] array) {
        return Unpooled.wrappedBuffer(array);
    }

    public static final ByteBuf wrappedBuffer(ByteBuf buf) {
        return Unpooled.wrappedBuffer(buf);
    }

    public static final ByteBuf wrappedBuffer(ByteBuffer buf) {
        return Unpooled.wrappedBuffer(buf);
    }

    public static final ByteBuf copiedBuffer(byte[] array) {
        return Unpooled.copiedBuffer(array);
    }

    public static final ByteBuf copiedBuffer(ByteBuf buf) {
        return Unpooled.copiedBuffer(buf);
    }

    public static final ByteBuf copiedBuffer(ByteBuffer buf) {
        return Unpooled.copiedBuffer(buf);
    }

    public static final ByteBuf copiedBuffer(CharSequence string, Charset charset) {
        return Unpooled.copiedBuffer(string, charset);
    }

    public static final ByteBuf createBuffer(int capacity) {
        return Unpooled.buffer(capacity);
    }

    public static final ByteBuf createDirectBuffer(int capacity) {
        return Unpooled.directBuffer(capacity);
    }

}

package arianrod.support;

import clojure.lang.IFn;
import clojure.lang.RT;
import clojure.lang.Var;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

import java.util.Set;


public class BootstrapFactory {


    public static ServerBootstrap createSimpleServerBootstrap(Class<? extends ServerChannel> channelClass, final IFn handler, final Set filter) {
        Var var = RT.var("arianrod.server", "*debug*");
        Boolean debug = (Boolean) var.get();
        var = RT.var("arianrod.server", "*backlog*");
        int backlog = ((Long) var.get()).intValue();

        ServerBootstrap b = new ServerBootstrap();
        b.channel(channelClass)
            .option(ChannelOption.SO_BACKLOG, backlog)
            .option(ChannelOption.TCP_NODELAY, true)
            .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new ChannelEventHandler(handler, filter));
                    }
                });

        if (debug) {
            b.handler(new LoggingHandler(LogLevel.DEBUG));
        }

        return b;
    }


    public static ServerBootstrap createServerBootstrap(Class<? extends ServerChannel> channelClass, final IFn initializer) {
        Var var = RT.var("arianrod.server", "*debug*");
        Boolean debug = (Boolean) var.get();
        var = RT.var("arianrod.server", "*backlog*");
        int backlog = ((Long) var.get()).intValue();

        ServerBootstrap b = new ServerBootstrap();
        b.channel(channelClass)
            .option(ChannelOption.SO_BACKLOG, (int) backlog)
            .option(ChannelOption.TCP_NODELAY, true)
            .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                        initializer.invoke(ch.pipeline());
                    }
                });

        if (debug) {
            b.handler(new LoggingHandler(LogLevel.DEBUG));
        }

        return b;
    }

    public static Bootstrap createBootstrap(Class<? extends Channel> channelClass, final IFn initializer) {
        Var var = RT.var("arianrod.client", "*debug*");
        Boolean debug = (Boolean) var.get();

        Bootstrap b = new Bootstrap();
        b.channel(channelClass)
            .option(ChannelOption.TCP_NODELAY, true)
            .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                        initializer.invoke(ch.pipeline());
                    }
                });

        if (debug) {
            b.handler(new LoggingHandler(LogLevel.DEBUG));
        }
        return b;
    }

    public static ChannelPipeline addLastPipeline(ChannelPipeline pipeline, String name, ChannelHandler handler) {
        return pipeline.addLast(name, handler);
    }

}

終わりに

素直に Go 書きましょう！