/**
 * 配置 SpringBoot WebSocket 支持
 *
 * @author rxliuli
 */
@Configuration
@EnableWebSocketMessageBroker
public class SpringWebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {
    /**
     * 注册一个 Socket 端点
     *
     * @param stompEndpointRegistry stomp 端点注册表
     */
    @Override
    public void registerStompEndpoints(StompEndpointRegistry stompEndpointRegistry) {
        stompEndpointRegistry.addEndpoint("/endpoint")
                //设置允许所有源请求（跨域）
                .setAllowedOrigins("*")
                .withSockJS();
    }

    /**
     * 注册一些广播消息代理
     *
     * @param registry 消息代理注册对象
     */
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        //注册简单代理（里面是前缀）
        //注：默认 topic 是主题（广播），user 则是用户（点对点）
        registry.enableSimpleBroker("/topic", "/user");
    }
}

/**
 * 双向广播控制器
 *
 * @author rxliuli
 */
@Controller
public class BilateralBroadcastingSocket {
    /**
     * 广播推送
     * 注解 @Payload 是为了绑定消息到参数 text 上
     *
     * @param text      简单的文本信息
     * @param sessionId 当前请求 socket 会话 id
     * @return 会话 id 和消息内容
     */
    @MessageMapping(value = "/talk")
    @SendTo("/topic/broadcasting/bilateral/allClient")
    public String talk(@Payload String text, @Header("simpSessionId") String sessionId) throws InterruptedException {
        //模拟处理其他事情
        Thread.sleep(1000L);
        return "[ " + sessionId + "] 说: [" + text + "]";
    }
}

/**
 * 单向广播控制器
 *
 * @author rxliuli
 */
@Controller
public class UnidirectionalBroadcastingSocket {
    /**
     * 从服务端推送消息到所有客户端
     * 这是单向推送到客户端的，不接受从客户端的输入
     */
    @SendTo("/topic/broadcasting/unidirectional/allClient")
    public Object broadcasting() {
        return null;
    }
}

/**
 * 使用 Scheduled 不停的推送信息
 *
 * @author rxliuli
 */
@Component
public class ScheduledRefreshJob {
    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;

    /**
     * 不停地推送消息到客户端
     */
    @Scheduled(fixedDelay = 10 * 1000)
    public void scheduledBroadcasting() {
        simpMessagingTemplate.convertAndSend("/topic/broadcasting/unidirectional/allClient", new Person(1L, "rx", false));
    }
}

/**
 * 双向点对点推送控制器
 *
 * @author rxliuli
 */
@Controller
public class BilateralPushSocket {
    /**
     * 点对点推送（双向通信）
     *
     * @param text      消息
     * @param sessionId 会话 id
     * @return 推送到当前会话的消息
     */
    @MessageMapping("/speak")
    @SendToUser("/push/bilateral/thisClient")
    public String speak(@Payload String text, @Header("simpSessionId") String sessionId) throws InterruptedException {
        //模拟处理其他事情
        Thread.sleep(1000L);
        return "[ " + sessionId + "] send: [" + text + "]";
    }
}

/**
 * 单向点对点推送服务端
 *
 * @author rxliuli
 */
@Controller
public class UnidirectionalPushSocket {
    /**
     * 从服务端推送消息到所有客户端
     * 这是单向推送到客户端的，不接受从客户端的输入
     */
    @SendToUser("/push/unidirectional/thisClient")
    public Object push() {
        return null;
    }
}

/**
 * 使用 Scheduled 不停的推送信息
 *
 * @author rxliuli
 */
@Component
public class ScheduledRefreshJob {
    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;

    /**
     * 不停推送消息到某个指定的客户端
     */
    @Scheduled(fixedDelay = 10 * 1000)
    public void scheduledPush() {
        simpMessagingTemplate.convertAndSendToUser("r2qspi4s", "/push/unidirectional/thisClient", new Person(2L, "琉璃", false));
    }
}

/**
* 用户 session 记录类
*
* @author rxliuli
*/
@Component
public class SocketSessionRegistry {
    /**
    * 未登录的用户默认存储的 user id
    */
    public static final String DIRECT_TOURIST = "DIRECT_TOURIST";
    /**
    * 这个集合存储 用户 id -> session 列表
    * 单个用户可能打开多个页面，就会出现多个 Socket 会话
    */
    private final ConcurrentMap<String, Set<String>> userSessionIds = new ConcurrentHashMap<>();
    private final Object lock = new Object();

    /**
    * 根据 user id 获取 sessionId
    *
    * @param user 用户 id
    * @return 用户关联的 sessionId
    */
    public Set<String> getSessionIds(String user) {
        Set<String> set = this.userSessionIds.get(user);
        return set != null ? set : Collections.emptySet();
    }

    /**
    * 获取所有 session
    *
    * @return 所有的 用户 id -> session 列表
    */
    public ConcurrentMap<String, Set<String>> getAllSessionIds() {
        return this.userSessionIds;
    }

    /**
    * 根据用户 id 注册一个 session
    *
    * @param user      用户 id
    * @param sessionId Socket 会话 id
    */
    public void registerSessionId(String user, String sessionId) {
        Assert.notNull(user, "User must not be null");
        Assert.notNull(sessionId, "Session ID must not be null");
        synchronized (this.lock) {
            Set<String> set = this.userSessionIds.get(user);
            if (set == null) {
                this.userSessionIds.put(user, new CopyOnWriteArraySet<>());
            }
            set.add(sessionId);
        }
    }

    /**
    * 根据用户 id 删除一个 session
    *
    * @param user      用户 id
    * @param sessionId Socket 会话 id
    */
    public void unregisterSessionId(String user, String sessionId) {
        Assert.notNull(user, "User Name must not be null");
        Assert.notNull(sessionId, "Session ID must not be null");
        synchronized (this.lock) {
            Set set = this.userSessionIds.get(user);
            if (set != null && set.remove(sessionId) && set.isEmpty()) {
                this.userSessionIds.remove(user);
            }
        }
    }
}

/**
* 会话事件监听基类
*
* @author rxliuli
*/
public abstract class BaseSessionEventListener<Event extends AbstractSubProtocolEvent> implements ApplicationListener<Event> {
    protected final Logger log = LoggerFactory.getLogger(getClass());
    @Autowired
    protected SocketSessionRegistry webAgentSessionRegistry;

    /**
    * 计算出 user id 和 session id 并传入到自定义的函数中
    *
    * @param event      事件
    * @param biConsumer 自定义的操作
    */
    protected void using(Event event, BiConsumer<String, String> biConsumer) {
        StompHeaderAccessor sha = StompHeaderAccessor.wrap(event.getMessage());
        //login get from browser
        List<String> shaNativeHeader = sha.getNativeHeader("Authorization");
        String user;
        if (shaNativeHeader == null || shaNativeHeader.isEmpty()) {
            user = null;
        } else {
            user = shaNativeHeader.get(0);
        }
        //如果当前用户没有登录（没有认证信息），就添加到游客里面
        if (user == null || "".equals(user) || "undefined".equals(user) || "null".equals(user)) {
            user = SocketSessionRegistry.DIRECT_TOURIST;
        }
        String sessionId = sha.getSessionId();
        biConsumer.accept(user, sessionId);
    }
}

/**
* Socket 连接建立监听
* 用于 session 注册 以及 key 值获取
*
* @author rxliuli
*/
@Component
public class SessionConnectEventListener extends BaseSessionEventListener<SessionConnectEvent> {
    @Override
    public void onApplicationEvent(SessionConnectEvent event) {
        using(event, (user, sessionId) -> webAgentSessionRegistry.registerSessionId(user, sessionId));
    }
}

/**
* Socket 会话断开监听
*
* @author rxliuli
*/
@Component
public class SessionDisconnectEventListener extends BaseSessionEventListener<SessionDisconnectEvent> {
    @Override
    public void onApplicationEvent(SessionDisconnectEvent event) {
        //这里先根据 session id 查询出 user，然后删除对应的会话 id
        //前端无法传递 token 到这里却是只能出此下策了
        using(event, (user, sessionId) -> webAgentSessionRegistry.getAllSessionIds().entrySet().stream()
                .filter(sse -> sse.getValue().contains(sessionId))
                .findFirst()
                .ifPresent(sse -> {
                    webAgentSessionRegistry.unregisterSessionId(sse.getKey(), sessionId);
                    log.info("Socket 连接断开，用户：{}，会话：{}", sse.getKey(), sessionId);
                }));
    }
}

stompClient.connect(getHeaders(), function(){
    console.log('打开 Socket 连接')
})
// TODO 这里还有一些问题，无法带上 headers 到后端
stompClient.disconnect(function () {
    console.log('断开连接');
}, getHeaders());

/**
   * 获取一个认证的 headers 信息
   * @return {{"X-Requested-With": string, Authorization: any}} 含有认证信息的 headers 对象
   */
  function getHeaders() {
      return {
          'X-Requested-With': 'X-Requested-With',
          'Authorization': localStorage.token
      }
  }

socketSessionRegistry.getAllSessionIds().entrySet().stream()
        .filter(kv -> !SocketSessionRegistry.DIRECT_TOURIST.equals(kv.getKey()))
        .forEach(kv -> kv.getValue().forEach(sessionId -> simpMessagingTemplate.convertAndSendToUser(sessionId, "/push/unidirectional/thisClient", new Person(2L, "琉璃", false))));

/**
 * 接受和发送复杂类型的消息
 *
 * @author rxliuli
 */
@Controller
public class ComplexMessageSocket {
    /**
     * 接收/返回复杂类型 Person 的对象
     *
     * @param person Person 类对象
     * @return Person 类对象
     */
    @MessageMapping("/complexMessage")
    @SendTo("/topic/complexMessage/allClient")
    public Person complexMessage(Person person) {
        return new Person().setName("Mr. " + person.getName());
    }
}

Exception in thread "main" java.lang.NoSuchMethodError: org.json.JSONArray.iterator()Ljava/util/Iterator;
	at com.aliyuncs.regions.LocalEndpointResolver.<init>(LocalEndpointResolver.java:39)
	at com.aliyuncs.profile.DefaultProfile.<init>(DefaultProfile.java:72)
	at com.aliyuncs.profile.DefaultProfile.getProfile(DefaultProfile.java:209)
	at com.rx.f3d.common.util.SmsSendUtil.main(SmsSendUtil.java:28)