Java 类io.netty.channel.ChannelId 实例源码

/**
 * 用户是否在线
 *
 * @param userId
 * @return
 */
boolean isOnline(long userId) {
    ChannelId channelId = LOGIN_USERS.inverse().get(userId);
    if (channelId == null) {
        return false;
    }
    Channel channel = CHANNEL_GROUP.find(channelId);
    if (channel == null || !channel.isActive()) {
        if (removeLock.tryLock()) {
            try {
                LOGIN_USERS.inverse().remove(userId);
            } finally {
                removeLock.unlock();
            }
        }
        return false;
    }
    return true;
}

/**
 * 获取用户的 channel
 *
 * @param userId
 * @return
 */
Channel getUserChannel(long userId) {
    ChannelId channelId = LOGIN_USERS.inverse().get(userId);
    if (channelId == null) {
        // 用户未登录
        throw new UserUnLoginException();
    }
    Channel channel = CHANNEL_GROUP.find(channelId);
    if (channel == null) {
        // 用户下线了
        throw new UserChannelNotFoundException();
    }
    if (!channel.isActive()) {
        this.channelInactive(channel);
        throw new UserChannelUnActiveException();
    }
    return channel;
}

protected void remove(Session session) {
    if (session == null) { return; }

    synchronized (this) {
        try {
            if (session.cleanSession()) { // [MQTT-3.1.2-5]
                sessions.remove(session.clientId());
            }

            ChannelId channelId = session.channelId();
            if (channelId == null) { return; }

            clientIds.remove(channelId);
            ctxs.remove(session.clientId());
        }
        finally {
            if (logger.isDebugEnabled()) {
                logger.debug("session removed [clientId={}, sessionsSize={}, clientIdsSize={}, ctxsSize={}]",
                        session.clientId(), sessions.size(), clientIds.size(), ctxs.size());
            }
        }
    }
}

private MqttConnAckMessage executeNormalChannelRead0(String clientId, boolean cleanSession, ChannelId channelId)
        throws Exception {
    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
            10);
    MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader("MQTT", 4, true, true, true, 0, true,
            cleanSession, 60);
    MqttConnectPayload payload = new MqttConnectPayload(clientId, "willtopic", "willmessage", "username",
            "password");

    MqttConnectMessage msg = new MqttConnectMessage(fixedHeader, variableHeader, payload);

    ChannelId cid = channelId == null ? TestUtil.newChannelId(clientId, false) : channelId;

    EmbeddedChannel channel = new EmbeddedChannel(cid, new ConnectReceiver());

    channel.writeInbound(msg);

    return channel.readOutbound();
}

@Override
public boolean remove(Object o) {
    Channel c = null;
    if (o instanceof ChannelId) {
        c = nonServerChannels.remove(o);
        if (c == null) {
            c = serverChannels.remove(o);
        }
    } else if (o instanceof Channel) {
        c = (Channel) o;
        if (c instanceof ServerChannel) {
            c = serverChannels.remove(c.id());
        } else {
            c = nonServerChannels.remove(c.id());
        }
    }

    if (c == null) {
        return false;
    }

    c.closeFuture().removeListener(remover);
    return true;
}

/**
 * 获取单例
 * @return
 */
public static SessionHelper instange(){
    synchronized (SessionHelper.class) {
        if (manager == null) {
            manager = new SessionHelper();
            if (manager.sessionMap == null) {
                // 需要线程安全的Map
                manager.sessionMap = new ConcurrentHashMap<ChannelId,HttpSession>();
            }
        }
    }
    return manager;
}

/**
 * 清除过期的session
 * 需要在定时器中执行该方法
 */
public void clearExpireSession(){
    Iterator<Map.Entry<ChannelId,HttpSession>> iterator = manager.sessionMap.entrySet().iterator();
    while(iterator.hasNext()){
        Map.Entry<ChannelId,HttpSession> sessionEntry = iterator.next();
        if(sessionEntry.getValue()==null || sessionEntry.getValue().isExpire()){
            iterator.remove();
        }
    }
}

public HttpSession(ChannelId id,ChannelHandlerContext context,Long createTime,Long expireTime){
    this.id = id;
    this.context = context;
    this.createTime = createTime;
    this.expireTime = expireTime;
    assertSessionMapNotNull();
}

/**
 * 根据channel id 获取用户登录的 user id
 *
 * @param channelId
 * @return
 */
Long getLoginUserId(ChannelId channelId) {
    if (CHANNEL_GROUP.find(channelId) == null) {
        return null;
    }
    return LOGIN_USERS.get(channelId);
}

public WebPlayerBridge(WebSocketServerThread webSocketServerThread, Settings settings) {
    this.webSocketServerThread = webSocketServerThread;
    this.setCustomNames = settings.setCustomNames;
    this.disableGravity = settings.disableGravity;
    this.disableAI = settings.disableAI;

    if (settings.entityClassName == null || "".equals(settings.entityClassName)) {
        this.entityClass = null;
    } else {
        try {
            this.entityClass = Class.forName("org.bukkit.entity." + settings.entityClassName);
        } catch (ClassNotFoundException ex) {
            ex.printStackTrace();

            // HumanEntity.class fails on Glowstone with https://gist.github.com/satoshinm/ebc87cdf1d782ba91b893fe24cd8ffd2
            // so use sheep instead for now. TODO: spawn ala GlowNPC: https://github.com/satoshinm/WebSandboxMC/issues/13
            webSocketServerThread.log(Level.WARNING, "No such entity class " + settings.entityClassName + ", falling back to Sheep");
            this.entityClass = Sheep.class;
        }
    }

    this.constrainToSandbox = settings.entityMoveSandbox;
    this.dieDisconnect = settings.entityDieDisconnect;

    this.clickableLinks = settings.clickableLinks;
    this.clickableLinksTellraw = settings.clickableLinksTellraw;
    this.publicURL = settings.publicURL;

    this.allowAnonymous = settings.allowAnonymous;
    this.lastPlayerID = 0;
    this.channelId2name = new HashMap<ChannelId, String>();
    this.channelId2Entity = new HashMap<ChannelId, Entity>();
    this.entityId2Username = new HashMap<Integer, String>();
    this.name2channel = new HashMap<String, Channel>();
}

public void broadcastLineExcept(ChannelId excludeChannelId, String message) {
    for (Channel channel: allUsersGroup) {
        if (channel.id().equals(excludeChannelId)) {
            continue;
        }

        channel.writeAndFlush(new BinaryWebSocketFrame(Unpooled.copiedBuffer((message + "\n").getBytes())));
    }
}

@JsonSerialize(using = ChannelIdSerializer.class)
@JsonProperty
public ChannelId channelId() {
    ChannelHandlerContext ctx = NEXUS.channelHandlerContext(clientId);
    if (ctx == null) { return null; }

    return ctx.channel().id();
}

@Override
public Channel find(ChannelId id) {
    Channel c = nonServerChannels.get(id);
    if (c != null) {
        return c;
    } else {
        return serverChannels.get(id);
    }
}

@Override
public boolean add(Channel channel) {
    ConcurrentMap<ChannelId, Channel> map =
        channel instanceof ServerChannel? serverChannels : nonServerChannels;

    boolean added = map.putIfAbsent(channel.id(), channel) == null;
    if (added) {
        channel.closeFuture().addListener(remover);
    }
    return added;
}

public HttpSession(ChannelId id,ChannelHandlerContext context){
    this(id,context,System.currentTimeMillis());
}

public HttpSession(ChannelId id,ChannelHandlerContext context,Long createTime){
    this(id,context,createTime,createTime + SessionConfig.instance().sessionTimeOut());
}

public ChannelId getId() {
    return id;
}

public void setId(ChannelId id) {
    this.id = id;
}

@Override
public ChannelId id() {
    return null;
}

private ChannelId channelId(Channel channel) {
    return Channels.getChannelId(channel);
}

public static ChannelId getChannelId(Channel channel) {
    Attribute<ChannelId> attr = channel.attr(CHANNEL_ID_ATTRIBUTE);
    return attr != null ? attr.get() : null;
}

public Map<ChannelId, String> getChannelIdUserIdMap() {
    return channelIdUserIdMap;
}

@Override
public void serialize(ChannelId value, JsonGenerator generator, SerializerProvider provider)
        throws IOException, JsonProcessingException {

    generator.writeString(value.toString());
}

public void dispose(boolean sendWill) {
    setConnected(false);

    if (sendWill && will != null) { // [MQTT-3.1.2-12]
        Topic topic = Topic.NEXUS.prepare(will);
        topic.publish(will);

        will(null); // [MQTT-3.1.2-10]
    }

    ChannelId channelId = null;
    ChannelHandlerContext ctx = NEXUS.channelHandlerContext(clientId);
    if (ctx != null) {
        ctx.channel().disconnect().addListener(ChannelFutureListener.CLOSE);
        channelId = ctx.channel().id();
    }

    logger.debug("Session disposed [clientId={}, channelId={}]", clientId, ctx == null ? "null" : channelId);

    EventExecutor executor = ctx != null ? ctx.channel().eventLoop() : GlobalEventExecutor.INSTANCE;
    executor.execute(
            () -> Plugins.INSTANCE.get(DisconnectEventListener.class).disconnected(new DisconnectEventArgs() {
                @Override
                public String clientId() {
                    return clientId;
                }

                @Override
                public Boolean cleanSession() {
                    return cleanSession;
                }

                @Override
                public Boolean byDisconnectMessage() {
                    return !sendWill;
                }
            }));

    // TODO WHY => Current thread is not owner of the lock! -> <not-locked>
    // disposeLock.lock();
    // try {
    if (cleanSession) {
        TopicSubscriber.NEXUS.removeByClientId(clientId);
        TopicSubscription.NEXUS.removeByClientId(clientId);
        OutboundMessageStatus.NEXUS.removeByClientId(clientId);
    }

    NEXUS.remove(this);
    // }
    // finally {
    // disposeLock.unlock();
    // }

    ClusterDataDisposer.INSTANCE.disposeLock(disposeLock);

}

@Override
public ChannelId id() {
    // TODO Auto-generated method stub
    return null;
}

public @Nullable Connection getConnection(@NotNull ChannelId channelId) {
    return connections.get(channelId);
}

@Override
public ChannelId id() {

    return null;
}

@Override
public ChannelId id() {
    return null;
}

/**
 * 접속 사용자 정보 제거
 *
 * @param channel Netty 채널
 */
public void removeUser(Channel channel) {

    ChannelId channelId = channel.id();
    Map<ChannelId, String> channelIdUserIdMap = channelIdUserIdRepository.getChannelIdUserIdMap();
    String userId = channelIdUserIdMap.get(channelId);

    // 사용자 정보 제거
    if (!StringUtils.isEmpty(userId)) {

        userIdChannelRepository.getUserIdChannelMap().remove(userId);

        String roomId = userIdRoomIdRepository.getUserIdRoomIdMap().get(userId);

        // 룸 정보 제거
        if (!StringUtils.isEmpty(roomId)) {

            roomIdUserIdRepository.getRoomIdUserIdMap().remove(roomId, userId);
            userIdRoomIdRepository.getUserIdRoomIdMap().remove(userId);

        }

        channelIdUserIdMap.remove(channelId);

    }

}

public Session get(ChannelId channelId) {
    String clientId = clientIds.get(channelId);

    if (Strings.isNullOrEmpty(clientId)) { return null; }

    return sessions.get(clientId);
}

/**
 * Returns the {@link Channel} which has the specified {@link ChannelId}.
 *
 * @return the matching {@link Channel} if found. {@code null} otherwise.
 */
Channel find(ChannelId id);