SpringB에서 RabbitMQ와 같은 외부 브로커와 함께 convertAndSendToUser()를 어떻게 사용합니까?

Question

나는 외부 브로커로 RabbitMQ를 사용하여 봄 4에서 웹 소켓 지원을 구성하려고 만하고 최대한 빨리 RabbitMQ로 전환, 나는 클라이언트의 시작에 다음과 같은 오류가 발생합니다 :

'/user/queue/changes' is not a valid destination. 
Valid destination types are: /temp-queue, /exchange, /topic, /queue, /amq/queue, /reply-queue/. 

에 서버 convertAndSendToUser 사용하고 간단한 브로커와 함께 잘 작동하지만 RabbitMQ로 전환하자마자이 오류가 발생합니다. RabbitMQ는 정상적인 주제의 브로드 캐스트에 잘 작동합니다. 단지 /user 채널이 넘어갑니다.

/user을 RabbitMQ와 함께 사용하려면 특별한 작업을 수행해야합니까?

편집

웹 소켓 구성을 포함하는 내 WebSocketConfig 몇 가지 사용자 정의한 꽤 표준 spring-session과 통합하는 것입니다

@Configuration 
@EnableWebSocketMessageBroker 
public class WebSocketConfig extends AbstractSessionWebSocketMessageBrokerConfigurer<ExpiringSession> { 


@Override 
public void configureStompEndpoints(StompEndpointRegistry registry) { 
    registry.addEndpoint("/changes").withSockJS(); 
} 

@Override 
public void configureMessageBroker(MessageBrokerRegistry config) { 
    //config.enableSimpleBroker("/queue", "/topic"); 
    StompBrokerRelayRegistration r = config.enableStompBrokerRelay("/user", "/topic"); 

    try { 
     String rabbitUrl = System.getenv("CLOUDAMQP_URL"); 
     if(rabbitUrl != null) { 
      log.info("RABBIT URL detected: " + rabbitUrl); 
      URI uri = new URI(rabbitUrl); 
      String host = uri.getHost();     
      String login = uri.getUserInfo().split(":",2)[0]; 
      String passCode = uri.getUserInfo().split(":",2)[1]; 
      String vhost = uri.getPath().substring(1); 
      r.setRelayHost(host); 
      r.setSystemLogin(login);  
      r.setSystemPasscode(passCode); 
      r.setClientLogin(login);  
      r.setClientPasscode(passCode); 
      r.setVirtualHost(vhost);     
     } 
    } catch(Exception e) { 
     log.error("Error setting up RabbitMQ", e); 
    } 
    config.setApplicationDestinationPrefixes("/app"); 
    } 
} 

Answer 1

나는 문제가 StompBrokerRelayRegistration의 R = 설정과 추측. enableStompBrokerRelay ("/ user", "/ topic") convertAndSendToUser.change StompBrokerRelayRegistration에서 "/ queue/loop-changes"를 ("/ queue", "/ topic")으로 지정합니다. 후속 질문이 있는데, Spring은 redditmq에 대한 destination에 sessionid를 추가 할 것입니다.이 sessionid는 DefaultUserSessionRegistry에서 가져온 것입니다 -> userSessionIds.s 클러스터에서 작동합니까?

Answer 2

나는 비슷한 오류를 만났지만 인터넷 검색 후 직관적 인 해결책을 찾지 못했습니다.

내 연구 결과를 RabbitMQ STOMP에 전달한 후 공유하고 싶습니다.

는 RabbitMQ STOMP의 문서에 따르면, /exchange, /queue, /amq/queue, /topic 및 temp-queue로 시작하는 경우에만 대상이 허용됩니다. 대상 /user/*은 허용되지 않습니다. 메시지 요구 사항에 따라 목적지를 선택할 수 있습니다.

/교환 - 임의의 라우팅 키로 보내고 SUBSCRIBE를 임의의 바인딩 패턴으로 보내십시오.

/queue - STOMP 게이트웨이가 관리하는 대기열에 대한 SEND 및 SUBSCRIBE.

/amq/queue - STOMP 게이트웨이 외부에서 생성 된 대기열에 대한 SEND 및 SUBSCRIBE;

/topic - 일시적인 주제와 지속 가능한 주제에 대한 SEND 및 SUBSCRIBE

/temp-queue/- 임시 대기열을 만듭니다 (회신 헤더 만).

예를 들어, 모든 구독자에게 알리기 위해 주제에 메시지를 보내고 싶습니다. 에 모든 활성 가입자를 각 메시지의 사본을 제공하는 간단한 주제 대상에 대한

은, 형태 /주제/< 이름>의 목적지 을 사용할 수 있습니다. 주제 대상은 AMQP 주제 교환의 모든 라우팅 패턴을 지원합니다.

활성 수신자가없는 주제 대상으로 보낸 메시지 은 단순히 버려집니다. SEND 용 프레임

AMQP 0-9-1 의미

는 메시지> 라우팅 키 < 이름으로 amq.topic 교환기로 전송된다. 프레임 구독 내용

, autodeleted, 비 영구 큐가 생성되고 키 < 이름> 라우팅과 amq.topic 교환 결합 . 큐에 대해 구독이 생성됩니다.

사양은 대상 /topic/<name>와 스톰 보조 메시지는 기본 교환 rabbitmq의 amp.topic를 사용하는 것을 의미합니다, 다음이 존재하지 않는 경우 변수 이름으로 만든 큐 될 수 바인딩 또한 해당 바인딩으로 교환 amp.topic을 바인드하도록 생성해야합니다.

영구 가입을 만들려면 클라이언트가 아래 헤더와 함께 가입 메시지를 보내야합니다. 내 응용 프로그램에서

durable:true 
auto-delete:false 

가 웹 소켓 서버가 메시지 /watch/{liveid} 수신 다음 주제 /topic/watchinfo-{liveid}에 다른 메시지를 회신 할 수 있습니다.

@Secured(User.ROLE_USER) 
@MessageMapping("/watch/{liveid}") 
@SendTo("/topic/watchinfo-{liveid}") 
@JsonView(View.Live.class) 
public LiveWatchInfoMessage liveinfo(@DestinationVariable("liveid") String liveid, 
             @AuthenticationPrincipal UserDetails activeUser) { 
     ... 
     return LiveWatchInfoMessage.builder().build(); 
    }