在数据库中修改某些内容时,仅通过WebSockets通知特定用户

Question

为了通过WebSockets通知所有用户,当在选定的JPA实体中修改某些内容时,我使用以下基本方法.

@ServerEndpoint("/Push")
public class Push {

    private static final Set<Session> sessions = new LinkedHashSet<Session>();

    @OnOpen
    public void onOpen(Session session) {
        sessions.add(session);
    }

    @OnClose
    public void onClose(Session session) {
        sessions.remove(session);
    }

    private static JsonObject createJsonMessage(String message) {
        return JsonProvider.provider().createObjectBuilder().add("jsonMessage", message).build();
    }

    public static void sendAll(String text) {
        synchronized (sessions) {
            String message = createJsonMessage(text).toString();

            for (Session session : sessions) {
                if (session.isOpen()) {
                    session.getAsyncRemote().sendText(message);
                }
            }
        }
    }
}

当修改选定的JPA实体时,会引发适当的CDI事件,该事件将由以下CDI观察者观察.

@Typed
public final class EventObserver {

    private EventObserver() {}

    public void onEntityChange(@Observes EntityChangeEvent event) {
        Push.sendAll("updateModel");
    }
}

观察者/使用者调用Push#sendAll()WebSockets端点中定义的静态方法,该方法将JSON消息作为通知发送给所有关联的用户/连接.

sendAll()当只有选定的用户被通知时,需要以某种方式修改方法内部的逻辑.

• 仅通知负责修改相关实体的用户(可能是管理员用户或只有在成功登录后才能修改内容的注册用户).
• 仅通知特定用户(不是全部)."特定"是指,例如,当在此站点上投票时,仅通知帖子所有者(该帖子可能由具有足够权限的任何其他用户投票).

建立初始握手时,HttpSession可以按照本答复中的说明进行访问,但仍然不足以完成上述两个子弹的任务.由于在第一次握手请求时可用,因此之后设置为该会话的任何属性在服务器端点中都不可用,换句话说,在建立握手之后设置的任何会话属性将不可用.

如上所述,仅通知所选用户的最可接受/规范方式是什么？需要方法中的某些条件语句sendAll()或其他某些条件语句.它似乎必须做除了用户之外的其他事情HttpSession.

我使用GlassFish Server 4.1/Java EE 7.

Answer 1

会议？

由于在第一次握手请求时可用,因此在服务器端点中将无法使用设置为该会话的任何属性,即换句话说,在建立握手后设置的任何会话属性将不可用

似乎你被"会话"这个词的含糊不清所困扰.会话的生命周期取决于上下文和客户端.websocket(WS)会话与HTTP会话的生命周期不同.就像EJB会话与HTTP会话的生命周期不同.就像传统的Hibernate会话与HTTP会话的生命周期不同.等等.您可能已经理解的HTTP会话在这里解释了servlet如何工作？实例化,会话,共享变量和多线程.这里解释了EJB会话JSF请求作用域bean是否继续在每个请求上重新创建新的有状态会话bean？

WebSocket生命周期

WS会话与HTML文档表示的上下文相关联.客户端基本上是JavaScript代码.WS会话在JavaScript开始时启动new WebSocket(url).当JavaScript明确调用实例close()上的函数时WebSocket,或者当关联的HTML文档因页面导航(单击链接/书签或修改浏览器地址栏中的URL),页面刷新或浏览器而被卸载时,WS会话停止标签/窗口关闭.请注意,您可以WebSocket在同一个DOM中创建多个实例,通常每个实例都有不同的URL路径或查询字符串参数.

每当WS会话开始时(即每次JavaScript执行时var ws = new WebSocket(url);),这将触发一个握手请求,您可以通过下面的Configurator类访问相关的HTTP会话,如您所知:

public class ServletAwareConfigurator extends Configurator {

    @Override
    public void modifyHandshake(ServerEndpointConfig config, HandshakeRequest request, HandshakeResponse response) {
        HttpSession httpSession = (HttpSession) request.getHttpSession();
        config.getUserProperties().put("httpSession", httpSession);
    }

}

因此,每个HTTP会话或HTML文档只会调用一次,就像您期望的那样.每次new WebSocket(url)创建时都会调用此方法.

然后@ServerEndpoint将创建带注释的类的全新实例,@OnOpen并将调用其带注释的方法.如果您熟悉JSF/CDI托管bean,只需将该类视为a @ViewScoped,将该方法视为a @PostConstruct.

@ServerEndpoint(value="/push", configurator=ServletAwareConfigurator.class)
public class PushEndpoint {

    private Session session;
    private EndpointConfig config;

    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        this.session = session;
        this.config = config;
    }

    @OnMessage
    public void onMessage(String message) {
        // ...
    }

    @OnError
    public void onError(Throwable exception) {
        // ...
    }

    @OnClose
    public void onClose(CloseReason reason) {
        // ...
    }

}

请注意,此类与servlet不同,而不是应用程序作用域.它基本上是WS会话作用域.所以每个新的WS会话都有自己的实例.这就是为什么你可以安全地分配Session和EndpointConfig作为实例变量.根据类设计(例如抽象模板等),您可以根据需要添加回Session所有其他onXxx方法的第一个参数.这也得到了支持.

该@OnMessageJavaScript时做注释的方法将被调用webSocket.send("some message").在@OnClose当WS会话关闭注释的方法会被调用.如有必要,确切的接近原因可由CloseReason.CloseCodesenum 提供的密切原因代码确定.在@OnError当一个异常被抛出,通常作为WS连接上IO错误(断管,连接复位等)注解的方法将被调用.

登录用户收集WS会话

回到仅通知特定用户的具体功能要求,您应该在上述说明之后理解您可以安全地依赖modifyHandshake()从关联的HTTP会话中提取登录用户,每次new WebSocket(url)都是在用户创建之后创建的登录.

public class UserAwareConfigurator extends Configurator {

    @Override
    public void modifyHandshake(ServerEndpointConfig config, HandshakeRequest request, HandshakeResponse response) {
        HttpSession httpSession = (HttpSession) request.getHttpSession();
        User user = (User) httpSession.getAttribute("user");
        config.getUserProperties().put("user", user);
    }

}

在带有a的WS端点类中@ServerEndpoint(configurator=UserAwareConfigurator.class),您可以在带@OnOpen注释的方法中获取它,如下所示:

@OnOpen
public void onOpen(Session session, EndpointConfig config) {
    User user = (User) config.getUserProperties().get("user");
    // ...
}

您应该在应用程序范围内收集它们.您可以static在端点类的字段中收集它们.或者,更好的是,如果在您的环境中没有破坏WS端点中的CDI支持(在WildFly中工作,而不是在Tomcat + Weld中,不确定GlassFish),那么只需在应用程序范围的CDI托管bean中收集它们,然后@Inject在端点类.

当User实例不是null(即用户登录时),请记住用户可以拥有多个WS会话.因此,您基本上需要在Map<User, Set<Session>>结构中收集它们,或者可能是更细粒度的映射,它们通过用户ID或组/角色映射它们,这毕竟允许更容易地找到特定用户.这一切都取决于最终的要求.这里至少是一个使用应用程序范围的CDI托管bean的启动示例:

@ApplicationScoped
public class PushContext {

    private Map<User, Set<Session>> sessions;

    @PostConstruct
    public void init() {
        sessions = new ConcurrentHashMap<>();
    }

    void add(Session session, User user) {
        sessions.computeIfAbsent(user, v -> ConcurrentHashMap.newKeySet()).add(session);
    }

    void remove(Session session) {
        sessions.values().forEach(v -> v.removeIf(e -> e.equals(session)));
    }

}

@ServerEndpoint(value="/push", configurator=UserAwareConfigurator.class)
public class PushEndpoint {

    @Inject
    private PushContext pushContext;

    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        User user = (User) config.getUserProperties().get("user");
        pushContext.add(session, user);
    }

    @OnClose
    public void onClose(Session session) {
        pushContext.remove(session);
    }

}

最后,您可以向特定用户发送消息,如下所示PushContext:

public void send(Set<User> users, String message) {
    Set<Session> userSessions;

    synchronized(sessions) {
        userSessions = sessions.entrySet().stream()
            .filter(e -> users.contains(e.getKey()))
            .flatMap(e -> e.getValue().stream())
            .collect(Collectors.toSet());
    }

    for (Session userSession : userSessions) {
        if (userSession.isOpen()) {
            userSession.getAsyncRemote().sendText(message);
        }
    }
}

该PushContext是一个CDI托管bean具有额外的优势,它是注射在任何其他CDI托管bean,用户可以方便地集成.

与关联用户一起开火CDI事件

在您EntityListener的CDI事件中,最有可能根据您之前的相关问题,使用JSF/Java EE从数据库进行实时更新,您已经拥有了已更改的实体,因此您应该能够找到与之相关的用户他们在模型中的关系.

仅通知负责修改相关实体的用户(可能是管理员用户或只有在成功登录后才能修改内容的注册用户)

@PostUpdate
public void onChange(Entity entity) {
    Set<User> editors = entity.getEditors();
    beanManager.fireEvent(new EntityChangeEvent(editors));
}

仅通知特定用户(不是全部)."特定"是指,例如,当在此站点上投票时,仅通知帖子所有者(该帖子可能由具有足够权限的任何其他用户投票).

@PostUpdate
public void onChange(Entity entity) {
    User owner = entity.getOwner();
    beanManager.fireEvent(new EntityChangeEvent(Collections.singleton(owner)));
}

然后在CDI事件观察员中,将其传递出去:

public void onEntityChange(@Observes EntityChangeEvent event) {
    pushContext.send(event.getUsers(), "message");
}

也可以看看:

• RFC6455 - WebSocket协议(描述ws://协议)
• W3 - WebSocket API(描述JS WebSocket接口)
• MDN - 编写WebSocket客户端应用程序(描述如何在客户端中使用WS API)
• Java EE 7教程 - WebSocket(描述javax.websocketAPI以及如何使用它)