聊聊RespServer

interface Resp {
  void channel(SocketChannel channel);
  void connected(ChannelHandlerContext ctx);
  void disconnected(ChannelHandlerContext ctx);
  void receive(ChannelHandlerContext ctx, RedisToken message);
}

public class RespServer implements Resp {

  private static final Logger LOGGER = LoggerFactory.getLogger(RespServer.class);

  private static final int BUFFER_SIZE = 1024 * 1024;
  private static final int MAX_FRAME_SIZE = BUFFER_SIZE * 100;

  private static final String DEFAULT_HOST = "localhost";
  private static final int DEFAULT_PORT = 12345;

  private EventLoopGroup bossGroup;
  private EventLoopGroup workerGroup;
  private ChannelFuture future;

  private final RespServerContext serverContext;

  public RespServer(RespServerContext serverContext) {
    this.serverContext = requireNonNull(serverContext);
  }

  public static Builder builder() {
    return new Builder();
  }

  public void start() {
    bossGroup = new NioEventLoopGroup();
    workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);

    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new RespInitializerHandler(this))
        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
        .option(ChannelOption.SO_RCVBUF, BUFFER_SIZE)
        .option(ChannelOption.SO_SNDBUF, BUFFER_SIZE)
        .childOption(ChannelOption.SO_KEEPALIVE, true)
        .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

    future = bootstrap.bind(serverContext.getHost(), serverContext.getPort());
    // Bind and start to accept incoming connections.
    future.syncUninterruptibly();

    serverContext.start();

    LOGGER.info("server started: {}:{}", serverContext.getHost(), serverContext.getPort());
  }

  public void stop() {
    try {
      if (future != null) {
        closeFuture(future.channel().close());
      }
      future = null;
    } finally {
      workerGroup = closeWorker(workerGroup);
      bossGroup = closeWorker(bossGroup);
    }

    serverContext.stop();

    LOGGER.info("server stopped");
  }

  @Override
  public void channel(SocketChannel channel) {
    LOGGER.debug("new channel: {}", sourceKey(channel));

    channel.pipeline().addLast("redisEncoder", new RedisEncoder());
    channel.pipeline().addLast("linDelimiter", new RedisDecoder(MAX_FRAME_SIZE));
    channel.pipeline().addLast(new RespConnectionHandler(this));
  }

  @Override
  public void connected(ChannelHandlerContext ctx) {
    String sourceKey = sourceKey(ctx.channel());
    LOGGER.debug("client connected: {}", sourceKey);
    getSession(ctx, sourceKey);
  }

  @Override
  public void disconnected(ChannelHandlerContext ctx) {
    String sourceKey = sourceKey(ctx.channel());

    LOGGER.debug("client disconnected: {}", sourceKey);

    serverContext.removeSession(sourceKey);
  }

  @Override
  public void receive(ChannelHandlerContext ctx, RedisToken message) {
    String sourceKey = sourceKey(ctx.channel());

    LOGGER.debug("message received: {}", sourceKey);

    parseMessage(message, getSession(ctx, sourceKey))
      .ifPresent(serverContext::processCommand);
  }

  //......

}

class RespInitializerHandler extends ChannelInitializer<SocketChannel> {

  private final Resp impl;

  RespInitializerHandler(Resp impl) {
    this.impl = impl;
  }

  @Override
  protected void initChannel(SocketChannel channel) throws Exception {
    impl.channel(channel);
  }

  @Override
  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    impl.disconnected(ctx);
  }
}

public class RedisEncoder extends MessageToByteEncoder<RedisToken> {
  @Override
  protected void encode(ChannelHandlerContext ctx, RedisToken msg, ByteBuf out) throws Exception {
    out.writeBytes(new RedisSerializer().encodeToken(msg));
  }
}

public class RedisDecoder extends ReplayingDecoder<Void> {

  private final int maxLength;

  public RedisDecoder(int maxLength) {
    this.maxLength = maxLength;
  }

  @Override
  protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
    out.add(parseResponse(buffer));
  }

  private RedisToken parseResponse(ByteBuf buffer) {
    RedisToken token = createParser(buffer).next();
    checkpoint();
    return token;
  }

  private RedisParser createParser(ByteBuf buffer) {
    return new RedisParser(maxLength, new NettyRedisSource(this, buffer));
  }

  //......

}

public class RespServerContext implements ServerContext {
  private static final Logger LOGGER = LoggerFactory.getLogger(RespServerContext.class);

  private final StateHolder state = new StateHolder();
  private final ConcurrentHashMap<String, Session> clients = new ConcurrentHashMap<>();
  private final Scheduler scheduler = Schedulers.from(Executors.newSingleThreadExecutor());

  private final String host;
  private final int port;
  private final CommandSuite commands;
  private SessionListener sessionListener;

  public RespServerContext(String host, int port, CommandSuite commands) {
    this(host, port, commands, nullListener());
  }

  public RespServerContext(String host, int port, CommandSuite commands,
                           SessionListener sessionListener) {
    this.host = requireNonNull(host);
    this.port = requireRange(port, 1024, 65535);
    this.commands = requireNonNull(commands);
    this.sessionListener = sessionListener;
  }

  public void start() {

  }

  public void stop() {
    clear();
    scheduler.shutdown();
  }

  @Override
  public int getClients() {
    return clients.size();
  }

  @Override
  public RespCommand getCommand(String name) {
    return commands.getCommand(name);
  }

  @Override
  public <T> Option<T> getValue(String key) {
    return state.getValue(key);
  }

  @Override
  public <T> Option<T> removeValue(String key) {
    return state.removeValue(key);
  }

  @Override
  public void putValue(String key, Object value) {
    state.putValue(key, value);
  }

  @Override
  public String getHost() {
    return host;
  }

  @Override
  public int getPort() {
    return port;
  }

  Session getSession(String sourceKey, Function<String, Session> factory) {
    return clients.computeIfAbsent(sourceKey, key -> {
      Session session = factory.apply(key);
      sessionListener.sessionCreated(session);
      return session;
    });
  }

  void processCommand(Request request) {
    LOGGER.debug("received command: {}", request);

    RespCommand command = getCommand(request.getCommand());
    try {
      executeOn(execute(command, request))
        .subscribe(response -> processResponse(request, response),
                   ex -> LOGGER.error("error executing command: " + request, ex));
    } catch (RuntimeException ex) {
      LOGGER.error("error executing command: " + request, ex);
    }
  }

  protected CommandSuite getCommands() {
    return commands;
  }

  protected void removeSession(String sourceKey) {
    Session session = clients.remove(sourceKey);
    if (session != null) {
      sessionListener.sessionDeleted(session);
    }
  }

  protected Session getSession(String key) {
    return clients.get(key);
  }

  protected RedisToken executeCommand(RespCommand command, Request request) {
    return command.execute(request);
  }

  protected <T> Observable<T> executeOn(Observable<T> observable) {
    return observable.observeOn(scheduler);
  }

  private void processResponse(Request request, RedisToken token) {
    request.getSession().publish(token);
    if (request.isExit()) {
      request.getSession().close();
    }
  }

  private Observable<RedisToken> execute(RespCommand command, Request request) {
    return Observable.create(observer -> {
      observer.onNext(executeCommand(command, request));
      observer.onComplete();
    });
  }

  private int requireRange(int value, int min, int max) {
    if (value <= min || value > max) {
      throw new IllegalArgumentException(min + " <= " + value + " < " + max);
    }
    return value;
  }

  private void clear() {
    clients.clear();
    state.clear();
  }
}

public interface RedisToken {
  RedisToken NULL_STRING = string((SafeString) null);
  RedisToken RESPONSE_OK = status("OK");

  RedisTokenType getType();

  <T> T accept(RedisTokenVisitor<T> visitor);

  static RedisToken nullString() {
    return NULL_STRING;
  }

  static RedisToken responseOk() {
    return RESPONSE_OK;
  }

  static RedisToken string(SafeString str) {
    return new StringRedisToken(str);
  }

  static RedisToken string(String str) {
    return new StringRedisToken(safeString(str));
  }

  static RedisToken status(String str) {
    return new StatusRedisToken(str);
  }

  static RedisToken integer(boolean b) {
    return new IntegerRedisToken(b ? 1 : 0);
  }

  static RedisToken integer(int i) {
    return new IntegerRedisToken(i);
  }

  static RedisToken error(String str) {
    return new ErrorRedisToken(str);
  }

  static RedisToken array(RedisToken... redisTokens) {
    return new ArrayRedisToken(ImmutableList.of(redisTokens));
  }

  static RedisToken array(Collection<RedisToken> redisTokens) {
    return new ArrayRedisToken(ImmutableList.from(redisTokens));
  }

  static RedisToken array(Sequence<RedisToken> redisTokens) {
    return new ArrayRedisToken(redisTokens.asArray());
  }

  static <T> Stream<T> visit(Stream<RedisToken> tokens, RedisTokenVisitor<T> visitor) {
    return tokens.map(token -> token.accept(visitor));
  }
}