Отправить уведомление отключенных упорным ТОПАЮТ очереди

Question

У меня есть этот код: на стороне клиента с JavaScript:

socket = new SockJS(context.backend + '/myWebSocketEndPoint'); 
    stompClient = Stomp.over(socket); 
    stompClient.connect({},function (frame) { 
      stompClient.subscribe('/queue/'+clientId+'/notification', function(response){ 
       alert(angular.fromJson(response.body)); 
      }); 
    }); 

В этом коде, клиент, когда соединяется, подписаться на получение уведомлений с помощью «/ очередью /» + его идентификатор клиента + '/ уведомление /. Поэтому у меня есть очередь для каждого клиента. Я использую stomp с sockjs

На моем сервере (Java + spring boot) у меня есть прослушиватель уведомлений, который при публикации события отправляет уведомление всем клиентам. Так у меня есть:

@Configuration 
@EnableWebSocketMessageBroker 
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer{ 

@Override 
    public void configureMessageBroker(MessageBrokerRegistry config) { 
    config.enableSimpleBroker("/queue"); 
} 


@Override 
public void registerStompEndpoints(StompEndpointRegistry registry) { 
    registry.addEndpoint("/myWebSocketEndPoint") 
      .setAllowedOrigins("*") 
      .withSockJS(); 
} 
} 

классового MenuItemNotificationChannel, называющее MenuItemNotificationSender отправить уведомление пользователей.

@Component 
public class MenuItemNotificationChannel extends AbstractNotificationChannel { 

@Autowired 
private MenuItemNotificationSender menuItemNotificationSender; 

@Autowired 
private UserRepository userRepository; 

@Override 
public void sendNotification(KitaiEvent<?> event, Map<String, Object> notificationConfiguration) throws Exception { 
    String menuItem = Optional.ofNullable((String) notificationConfiguration.get(MENU_ENTRY_KEY)).orElseThrow(IllegalArgumentException::new); 
    List<User> userList = userRepository.findAll(); 
    for(User u: userList){ 
     menuItemNotificationSender.sendNotification(new MenuItemDto(menuItem),u.getId()); 
    } 

MenuItemNotificationSender класс:

@Component 
public class MenuItemNotificationSender { 

@Autowired 
private SimpMessagingTemplate messagingTemplate; 

@Autowired 
public MenuItemNotificationSender(SimpMessagingTemplate messagingTemplate){ 
    this.messagingTemplate = messagingTemplate; 
} 

public void sendNotification(MenuItemDto menuItem,Long id) { 
    String address = "/queue/"+id+"/notification"; 
    messagingTemplate.convertAndSend(address, menuItem); 
} 
} 

Этот код работает отлично: уведомления отправляются каждому пользователю. Но если пользователь не подключен к сети, оповещения теряются. Мои вопросы:

• Как я могу проверить whit stomp какие подписи активны, а какие нет? (Если я могу проверить, активна ли подписка, я решаю свою проблему, потому что я сохраняю уведомление для пользователей в автономном режиме, а затем отправляю их при входе в систему)

• Могу ли я использовать постоянные очереди? (Я читал что-то об этом, но я не понимаю, если я могу использовать его только с топать и sockjs)

Извините за мой английский! : D

Answer 1

Вы можете поставить слушателя весной событие на сессии связано событие и событие сеанса отключения я тестировал этот с пружиной 4.3.4

@Component 
public class WebSocketSessionListener 
{ 
    private static final Logger logger = LoggerFactory.getLogger(WebSocketSessionListener.class.getName()); 
    private List<String> connectedClientId = new ArrayList<String>(); 

    @EventListener 
    public void connectionEstablished(SessionConnectedEvent sce) 
    { 
     MessageHeaders msgHeaders = sce.getMessage().getHeaders(); 
     Principal princ = (Principal) msgHeaders.get("simpUser"); 
     StompHeaderAccessor sha = StompHeaderAccessor.wrap(sce.getMessage()); 
     List<String> nativeHeaders = sha.getNativeHeader("userId"); 
     if(nativeHeaders != null) 
     { 
      String userId = nativeHeaders.get(0); 
      connectedClientId.add(userId); 
      if(logger.isDebugEnabled()) 
      { 
       logger.debug("Connessione websocket stabilita. ID Utente "+userId); 
      } 
     } 
     else 
     { 
      String userId = princ.getName(); 
      connectedClientId.add(userId); 
      if(logger.isDebugEnabled()) 
      { 
       logger.debug("Connessione websocket stabilita. ID Utente "+userId); 
      } 
     } 
    } 

    @EventListener 
    public void webSockectDisconnect(SessionDisconnectEvent sde) 
    { 
     MessageHeaders msgHeaders = sde.getMessage().getHeaders(); 
     Principal princ = (Principal) msgHeaders.get("simpUser"); 
     StompHeaderAccessor sha = StompHeaderAccessor.wrap(sde.getMessage()); 
     List<String> nativeHeaders = sha.getNativeHeader("userId"); 
     if(nativeHeaders != null) 
     { 
      String userId = nativeHeaders.get(0); 
      connectedClientId.remove(userId); 
      if(logger.isDebugEnabled()) 
      { 
       logger.debug("Disconnessione websocket. ID Utente "+userId); 
      } 
     } 
     else 
     { 
      String userId = princ.getName(); 
      connectedClientId.remove(userId); 
      if(logger.isDebugEnabled()) 
      { 
       logger.debug("Disconnessione websocket. ID Utente "+userId); 
      } 
     } 
    } 

    public List<String> getConnectedClientId() 
    { 
     return connectedClientId; 
    } 
    public void setConnectedClientId(List<String> connectedClientId) 
    { 
     this.connectedClientId = connectedClientId; 
    } 
} 

Когда клиент подключен добавить в список клиентов id идентификатора клиента; когда он отключается, вы удалите его

Затем вы можете ввести этот компонент или его список, где вы хотите проверить, активен ли клиент или меньше, и затем вы можете проверить, находится ли идентификатор клиента между идентификатором подключенных клиентов, который вы можете отправить сообщение, в противном случае вы должны сохранить его и отправить позже

на стороне клиента вы можете сделать что-то вроде этого:

var socket = new SockJS('/ws'); 
stompClient = Stomp.over(socket); 
stompClient.connect({userId:"customUserId"}, function (frame) { 
}); 

Angelo

Answer 2

почему бы не использовать некоторые события, как ниже, вы можете экспортировать классы дифферент файлы и использование SessionConnectedEvent и SessionDisconnectEvent OR SessionSubscribeEvent и SessionUnsubscribeEvent. см Doc здесь http://docs.spring.io/spring/docs/current/spring-framework-reference/html/websocket.html#websocket-stomp-appplication-context-events

import java.util.Collections; 
import java.util.LinkedList; 
import java.util.List; 

import org.springframework.context.ApplicationListener; 
import org.springframework.stereotype.Component; 
import org.springframework.web.socket.messaging.SessionConnectedEvent; 
import org.springframework.web.socket.messaging.SessionDisconnectEvent; 
import org.springframework.web.socket.messaging.SessionSubscribeEvent; 
import org.springframework.web.socket.messaging.SessionUnsubscribeEvent; 

@Component 
public class SessionConnectedListener extends SessionsListener implements ApplicationListener<SessionConnectedEvent> { 

    @Override 
    public void onApplicationEvent(SessionConnectedEvent event) { 
     users.add(event.getUser().getName()); 
    } 

} 

@Component 
class SessionDisconnectListener extends SessionsListener implements ApplicationListener<SessionDisconnectEvent> { 

    @Override 
    public void onApplicationEvent(SessionDisconnectEvent event) { 
     users.remove(event.getUser().getName()); 
    } 
} 

@Component 
class SessionSubscribeListener extends SessionsListener implements ApplicationListener<SessionSubscribeEvent> { 

    @Override 
    public void onApplicationEvent(SessionSubscribeEvent event) { 
     users.add(event.getUser().getName()); 
    } 
} 

@Component 
class SessionUnsubscribeListener extends SessionsListener implements ApplicationListener<SessionUnsubscribeEvent> { 

    @Override 
    public void onApplicationEvent(SessionUnsubscribeEvent event) { 
     users.remove(event.getUser().getName()); 
    } 
} 

class SessionsListener { 

    protected List<String> users = Collections.synchronizedList(new LinkedList<String>()); 

    public List<String> getUsers() { 
     return users; 
    } 
} 

и изменить код:

@Autowired 
private SessionsListener sessionsListener; 

@Override 
public void sendNotification(KitaiEvent<?> event, Map<String, Object> notificationConfiguration) throws Exception { 
    String menuItem = Optional.ofNullable((String) notificationConfiguration.get(MENU_ENTRY_KEY)).orElseThrow(IllegalArgumentException::new); 
    List<String> userList = sessionsListener.getUsers(); 
    for(String u: userList){ 
     menuItemNotificationSender.sendNotification(new MenuItemDto(menuItem),u); 
    }