聊聊claudb的pubsub command

@Command("publish")
@ParamLength(2)
public class PublishCommand implements DBCommand, SubscriptionSupport, PatternSubscriptionSupport {
​
  @Override
  public RedisToken execute(Database db, Request request) {
    String channel = request.getParam(0).toString();
    SafeString message = request.getParam(1);
    return integer(publishAll(getClauDB(request.getServerContext()), channel, message));
  }
​
  private int publishAll(DBServerContext server, String channel, SafeString message) {
    int count = publish(server, channel, message);
    int pcount = patternPublish(server, channel, message);
    return count + pcount;
  }
}

@ReadOnly
@Command("subscribe")
@ParamLength(1)
@PubSubAllowed
public class SubscribeCommand implements DBCommand, SubscriptionSupport {
​
  private static final String SUBSCRIBE = "subscribe";
​
  @Override
  public RedisToken execute(Database db, Request request) {
    Database admin = getAdminDatabase(request.getServerContext());
    String sessionId = getSessionId(request);
    Sequence<SafeString> channels = getChannels(request);
    int i = channels.size();
    List<Object> result = new LinkedList<>();
    for (SafeString channel : request.getParams()) {
      addSubscription(admin, sessionId, channel);
      getSessionState(request.getSession()).addSubscription(channel);
      result.addAll(asList(SUBSCRIBE, channel, ++i));
    }
    return convert(result);
  }
​
  private String getSessionId(Request request) {
    return request.getSession().getId();
  }
​
  private Sequence<SafeString> getChannels(Request request) {
    return getSessionState(request.getSession()).getSubscriptions();
  }
}

@ReadOnly
@Command("unsubscribe")
@ParamLength(1)
@PubSubAllowed
public class UnsubscribeCommand implements DBCommand, SubscriptionSupport {
​
  private static final String UNSUBSCRIBE = "unsubscribe";
​
  @Override
  public RedisToken execute(Database db, Request request) {
    Database admin = getAdminDatabase(request.getServerContext());
    String sessionId = getSessionId(request);
    Sequence<SafeString> channels = getChannels(request);
    int i = channels.size();
    List<Object> result = new LinkedList<>();
    for (SafeString channel : request.getParams()) {
      removeSubscription(admin, sessionId, channel);
      getSessionState(request.getSession()).removeSubscription(channel);
      result.addAll(asList(UNSUBSCRIBE, channel, --i));
    }
    return convert(result);
  }
​
  private String getSessionId(Request request) {
    return request.getSession().getId();
  }
​
  private Sequence<SafeString> getChannels(Request request) {
    return getSessionState(request.getSession()).getSubscriptions();
  }
}

@ReadOnly
@Command("psubscribe")
@ParamLength(1)
@PubSubAllowed
public class PatternSubscribeCommand implements DBCommand, PatternSubscriptionSupport {
​
  private static final String PSUBSCRIBE = "psubscribe";
​
  @Override
  public RedisToken execute(Database db, Request request) {
    Database admin = getAdminDatabase(request.getServerContext());
    String sessionId = getSessionId(request);
    Sequence<SafeString> channels = getChannels(request);
    int i = channels.size();
    List<Object> result = new LinkedList<>();
    for (SafeString pattern : request.getParams()) {
      addPatternSubscription(admin, sessionId, pattern);
      getSessionState(request.getSession()).addSubscription(pattern);
      result.addAll(asList(PSUBSCRIBE, pattern, ++i));
    }
    return convert(result);
  }
​
  private String getSessionId(Request request) {
    return request.getSession().getId();
  }
​
  private Sequence<SafeString> getChannels(Request request) {
    return getSessionState(request.getSession()).getSubscriptions();
  }
}

@ReadOnly
@Command("punsubscribe")
@ParamLength(1)
@PubSubAllowed
public class PatternUnsubscribeCommand implements DBCommand, PatternSubscriptionSupport {
​
  private static final String PUNSUBSCRIBE = "punsubscribe";
​
  @Override
  public RedisToken execute(Database db, Request request) {
    Database admin = getAdminDatabase(request.getServerContext());
    String sessionId = getSessionId(request);
    Sequence<SafeString> channels = getChannels(request);
    int i = channels.size();
    List<Object> result = new LinkedList<>();
    for (SafeString channel : request.getParams()) {
      removePatternSubscription(admin, sessionId, channel);
      getSessionState(request.getSession()).removeSubscription(channel);
      result.addAll(asList(PUNSUBSCRIBE, channel, --i));
    }
    return convert(result);
  }
​
  private String getSessionId(Request request) {
    return request.getSession().getId();
  }
​
  private Sequence<SafeString> getChannels(Request request) {
    return getSessionState(request.getSession()).getSubscriptions();
  }
}