@MessageMapping("connect.echo-channel")
void echoChannel(RSocketRequester requester) {
	runTest(() -> {
		Flux<String> flux = requester.route("echo-channel")
				.data(Flux.range(1, 10).map(i -> "Hello " + i), String.class)
				.retrieveFlux(String.class);

		StepVerifier.create(flux)
				.expectNext("Hello 1 async")
				.expectNextCount(7)
				.expectNext("Hello 9 async")
				.expectNext("Hello 10 async")
				.thenCancel()  // https://github.com/rsocket/rsocket-java/issues/613
				.verify(Duration.ofSeconds(5));
	});
}
 

@MessageMapping("/create/empty")
@SendToUser(broadcast = false)
public int createEmptyPlaylist(Principal p) {
    Locale locale = localeResolver.resolveLocale(p.getName());
    DateTimeFormatter dateFormat = DateTimeFormatter.ofLocalizedDateTime(FormatStyle.MEDIUM, FormatStyle.SHORT).withLocale(locale);

    Instant now = Instant.now();
    Playlist playlist = new Playlist();
    playlist.setUsername(p.getName());
    playlist.setCreated(now);
    playlist.setChanged(now);
    playlist.setShared(false);
    playlist.setName(dateFormat.format(now.atZone(ZoneId.systemDefault())));

    playlistService.createPlaylist(playlist);

    return playlist.getId();
}
 

@MessageMapping("connect.echo-stream")
void echoStream(RSocketRequester requester) {
	runTest(() -> {
		Flux<String> flux = requester.route("echo-stream").data("Hello").retrieveFlux(String.class);

		StepVerifier.create(flux)
				.expectNext("Hello 0")
				.expectNextCount(5)
				.expectNext("Hello 6")
				.expectNext("Hello 7")
				.thenCancel()
				.verify(Duration.ofSeconds(5));
	});
}
 

@MessageMapping("/message/{foo}/{name}")
public void messageMappingDestinationVariable(@DestinationVariable("foo") String param1,
		@DestinationVariable("name") String param2) {
	this.method = "messageMappingDestinationVariable";
	this.arguments.put("foo", param1);
	this.arguments.put("name", param2);
}
 

@MessageMapping("/chat")
//@SendTo("/topic")
//@SendToUser()
public void chatEndpoint(@Payload WsMessage wsMessage) {
    System.out.println(wsMessage);
    messagingTemplate.convertAndSend("/topic", wsMessage);
}
 

@MessageMapping("/chat")
public void handleChat(Principal principal,String msg){
    // 在 SpringMVC 中，可以直接在参数中获得 principal，principal 中包含当前用户信息
    if (principal.getName().equals("nasus")){
        // 硬编码，如果发送人是 nasus 则接收人是 chenzy 反之也成立。
        // 通过 messageingTemplate.convertAndSendToUser 方法向用户发送信息，参数一是接收消息用户，参数二是浏览器订阅地址，参数三是消息本身
        messagingTemplate.convertAndSendToUser("chenzy",
                "/queue/notifications",principal.getName()+"-send:" + msg);
    } else {
        messagingTemplate.convertAndSendToUser("nasus",
                "/queue/notifications",principal.getName()+"-send:" + msg);
    }
}
 

@MessageMapping("/update")
public void updatePlaylist(PlaylistUpdateRequest req) {
    Playlist playlist = new Playlist(playlistService.getPlaylist(req.getId()));
    playlist.setName(req.getName());
    playlist.setComment(req.getComment());
    playlist.setShared(req.getShared());
    playlistService.updatePlaylist(playlist);
}
 

/**
 * This @MessageMapping is intended to be used "stream <--> stream" style.
 * The incoming stream contains the interval settings (in seconds) for the outgoing stream of messages.
 *
 * @param settings
 * @return
 */
@PreAuthorize("hasRole('USER')")
@MessageMapping("channel")
Flux<Message> channel(final Flux<Duration> settings, @AuthenticationPrincipal UserDetails user) {
    log.info("Received channel request...");
    log.info("Channel initiated by '{}' in the role '{}'", user.getUsername(), user.getAuthorities());

    return settings
            .doOnNext(setting -> log.info("Channel frequency setting is {} second(s).", setting.getSeconds()))
            .doOnCancel(() -> log.warn("The client cancelled the channel."))
            .switchMap(setting -> Flux.interval(setting)
                    .map(index -> new Message(SERVER, CHANNEL, index)));
}
 

@MessageMapping("/files/moveup")
@SendToUser(broadcast = false)
public int up(PlaylistFilesModificationRequest req) {
    // in this context, modifierIds has one element that is the index of the file
    List<MediaFile> files = playlistService.getFilesInPlaylist(req.getId(), true);
    if (req.getModifierIds().size() == 1 && req.getModifierIds().get(0) > 0) {
        Collections.swap(files, req.getModifierIds().get(0), req.getModifierIds().get(0) - 1);
        playlistService.setFilesInPlaylist(req.getId(), files);
    }

    return req.getId();
}
 

@MessageMapping("/files/movedown")
@SendToUser(broadcast = false)
public int down(PlaylistFilesModificationRequest req) {
    // in this context, modifierIds has one element that is the index of the file
    List<MediaFile> files = playlistService.getFilesInPlaylist(req.getId(), true);
    if (req.getModifierIds().size() == 1 && req.getModifierIds().get(0) < files.size() - 1) {
        Collections.swap(files, req.getModifierIds().get(0), req.getModifierIds().get(0) + 1);
        playlistService.setFilesInPlaylist(req.getId(), files);
    }

    return req.getId();
}
 

@MessageMapping("/files/rearrange")
@SendToUser(broadcast = false)
public int rearrange(PlaylistFilesModificationRequest req) {
    // in this context, modifierIds are indices
    List<MediaFile> files = playlistService.getFilesInPlaylist(req.getId(), true);
    MediaFile[] newFiles = new MediaFile[files.size()];
    for (int i = 0; i < req.getModifierIds().size(); i++) {
        newFiles[i] = files.get(req.getModifierIds().get(i));
    }
    playlistService.setFilesInPlaylist(req.getId(), Arrays.asList(newFiles));

    return req.getId();
}
 

@MessageMapping("/remove")
public DownloadList remove(DownloadList download) {
    if (downloadService.remove(download) >= 0) {
        return download;
    } else {
        return null;
    }
}
 

@MessageMapping("/chat.sendMessage")
public void sendMessage(@Payload ChatMessage chatMessage) {
    try {
        redisTemplate.convertAndSend(msgToAll, JsonUtil.parseObjToJson(chatMessage));
    } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
    }
}
 

@MessageMapping("/chat.addUser")
public void addUser(@Payload ChatMessage chatMessage, SimpMessageHeaderAccessor headerAccessor) {

    LOGGER.info("User added in Chatroom:" + chatMessage.getSender());
    try {
        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
        redisTemplate.opsForSet().add(onlineUsers, chatMessage.getSender());
        redisTemplate.convertAndSend(userStatus, JsonUtil.parseObjToJson(chatMessage));
    } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
    }
}
 

/**
 * 聊天室发布订阅
 *
 * @param messageRO 消息请求对象
 * @param user 发送消息的用户对象
 * @throws Exception
 */
@MessageMapping(StompConstant.PUB_CHAT_ROOM)
public void chatRoom(MessageRO messageRO, User user) throws Exception {
    String message = messageRO.getMessage();

    if (!CheckUtils.checkMessageRo(messageRO) || !CheckUtils.checkUser(user)) {
        throw new ErrorCodeException(CodeEnum.INVALID_PARAMETERS);
    }
    if (CheckUtils.checkMessage(message) && message.startsWith(RobotConstant.prefix)) {
        messageService.sendMessageToRobot(StompConstant.SUB_CHAT_ROOM, message, user);
    }

    messageService.sendMessage(StompConstant.SUB_CHAT_ROOM, new MessageVO(user, message, messageRO.getImage(),
            MessageTypeEnum.USER));
}
 

@MessageMapping("/start")
public void start(@DestinationVariable int playerId, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.start(player);
}
 

@MessageMapping("/stop")
public void stop(@DestinationVariable int playerId, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.stop(player);
}
 

@MessageMapping("/exception")
public void handleWithError() {
	throw new IllegalArgumentException("Bad input");
}
 

@MessageMapping("/validation/payload")
public void payloadValidation(@Validated @Payload String payload) {
	this.method = "payloadValidation";
	this.arguments.put("message", payload);
}
 

@MessageMapping("/reloadsearch")
public void reloadSearchCriteria(@DestinationVariable int playerId, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.reloadSearchCriteria(player, headers.getSessionId());
}
 

@MessageMapping("/save")
@SendToUser
public int savePlayQueue(@DestinationVariable int playerId, PlayQueueRequest req, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    return playQueueService.savePlayQueue(player, req.getIndex(), req.getOffset());
}
 

@MessageMapping("/play/saved")
public void loadSavedPlayQueue(@DestinationVariable int playerId, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.loadSavedPlayQueue(player, headers.getSessionId());
}
 

@MessageMapping("/play/mediafile")
public void playMediaFile(@DestinationVariable int playerId, PlayQueueRequest req, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.playMediaFile(player, req.getId(), headers.getSessionId());
}
 

@MessageMapping("/play/playlist")
public void playPlaylist(@DestinationVariable int playerId, PlayQueueRequest req, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.playPlaylist(player, req.getId(), req.getIndex(), headers.getSessionId());
}
 

@MessageMapping("/play/topsongs")
public void playTopSong(@DestinationVariable int playerId, PlayQueueRequest req, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.playTopSong(player, req.getId(), req.getIndex(), headers.getSessionId());
}
 

@MessageMapping("/foo")
@SendTo("/bar")
public String handleMessage() {
	return "bar";
}
 

@MessageMapping("hello")
public Mono<String> hello(String name) {
	return Mono.just("Hello " + name);
}
 

@MessageMapping("A.*")
void ambiguousMatchA(String payload) {
	throw new IllegalStateException("Unexpected call");
}
 

@MessageMapping("mono")
public Mono<String> handleMono() {
	this.mono = MonoProcessor.create();
	return this.mono;
}
 

@MessageMapping("/play/shuffle")
public void playShuffle(@DestinationVariable int playerId, PlayQueueRequest req, SimpMessageHeaderAccessor headers) throws Exception {
    Player player = getPlayer(playerId, headers);
    playQueueService.playShuffle(player, req.getAlbumListType(), (int) req.getOffset(), req.getCount(),
            req.getGenre(), req.getDecade(), headers.getSessionId());
}