vlambda博客
学习文章列表

    @StreamListener 
    @Output(Processor.OUTPUT) 
    public Flux<Comment> save(@Input(Processor.INPUT) Flux<Comment> 
     newComment) { 
       return repository 
        .saveAll(newComment) 
        .map(comment -> { 
          log.info("Saving new comment " + comment); 
          meterRegistry
            .counter("comments.consumed", "imageId", comment.getImageId())
            .increment(); 
          return comment; 
        }); 
    } 

    server: 
      port: 9000 
 
    spring: 
      cloud: 
        stream: 
          bindings: 
            input: 
              destination: learning-spring-boot-comments 
              group: comments-service 
              content-type: application/json 
            output: 
              destination: learning-spring-boot-chatgroup: comments-chatcontent-type: application/json 

2017-07-05 00:00:36.769  INFO 92207 --- [ments-service-1] 
c.g.l.comments.CommentService : Saving new comment Comment(id=null,
imageId=581d6669596aec65dc9e6c05, comment=Nice cover!)

    compile('org.springframework.boot:spring-boot-starter-webflux') 
    compile('org.projectlombok:lombok') 
    compile('org.springframework.cloud:spring-cloud-starter-stream-
     rabbit') 
    compile('org.springframework.cloud:spring-cloud-stream-reactive') 
    compile('org.springframework.cloud:spring-cloud-starter-eureka') 
    compile('org.springframework.cloud:spring-cloud-starter-config') 

    @SpringCloudApplication 
    @EnableEurekaClient 
    public class LearningSpringBootChatApplication { 
 
      public static void main(String[] args) { 
        SpringApplication.run( 
          LearningSpringBootChatApplication.class, args); 
      } 
    } 

    spring: 
      application: 
        name: chat 

    server: 
      port: 8200 
 
    spring: 
      cloud: 
        stream: 
          bindings: 
            input: 
              destination: learning-spring-boot-chat 
              group: comments-chat 
              content-type: application/json 
            newComments: 
              destination: learning-spring-boot-chat 
              group: comments-chat 
              content-type: application/json 
            clientToBroker: 
              destination: learning-spring-boot-chat-user-messages 
              group: app-chatMessages 
            brokerToClient: 
              destination: learning-spring-boot-chat-user-messages 
              group: topic-chatMessages 

    <script th:inline="javascript"> /*<![CDATA[*/ (function() { ... custom JavaScript code here... })(); /*]]>*/ </script> 

    var socket = new WebSocket(
      'ws://localhost:8200/topic/comments.new'); 
    socket.onopen = function(event) { 
      console.log('Connected to chat service!'); 
      console.log(event); 
    } 
    socket.onmessage = function(event) { 
      console.log('Received ' + event.data + '!'); 
      var parsedMessage = JSON.parse(event.data); 
      var ul = document.getElementById( 
        'comments-' + parsedMessage.imageId); 
      var li = document.createElement('li'); 
      li.appendChild( 
        document.createTextNode(parsedMessage.comment)); 
      ul.appendChild(li); 
    } 

    @PostMapping("/comments") 
    public Mono<String> addComment(Mono<Comment> newComment) { 
 
      /* stream comments to COMMENTS service */ 
 
      return Mono.just("redirect:/"); 
    } 

    <td> 
      <input th:id="'comment-' + ${image.id}" type="text" value="" /> 
      <button th:id="${image.id}" class="comment">Submit</button> 
    </td> 

    // Register a handler for each button to make an AJAX call 
    document.querySelectorAll('button.comment') 
     .forEach(function(button) { 
       button.addEventListener('click', function() { 
         var comment = document.getElementById( 
           'comment-' + button.id); 
 
         var xhr = new XMLHttpRequest(); 
         xhr.open('POST', /*[[@{'/comments'}]]*/'', true); 
 
         var formData = new FormData(); 
         formData.append('comment', comment.value); 
         formData.append('imageId', button.id); 
 
         xhr.send(formData); 
 
         comment.value = ''; 
       }); 
    }); 

    @Controller 
    @EnableBinding(Source.class) 
    public class CommentController { 
      ... 
    } 

    @RestController 
    @EnableBinding(Source.class) 
    public class CommentController { 
      ... 
    } 

    @PostMapping("/comments") 
    public Mono<ResponseEntity<?>> addComment(Mono<Comment> newComment)
    { 
      if (commentSink != null) { 
        return newComment 
         .map(comment -> { 
           commentSink.next(MessageBuilder 
            .withPayload(comment) 
            .setHeader(MessageHeaders.CONTENT_TYPE, 
             MediaType.APPLICATION_JSON_VALUE) 
             .build()); 
           return comment; 
         }) 
         .flatMap(comment -> { 
           meterRegistry
             .counter("comments.produced", "imageId", comment.getImageId())
             .increment(); 
           return Mono.just(ResponseEntity.noContent().build()); 
         }); 
      } else { 
          return Mono.just(ResponseEntity.noContent().build()); 
      } 
    } 

    <div id="chatBox"> 
        Greetings! 
        <br/> 
        <textarea id="chatDisplay" rows="10" cols="80" disabled="true"></textarea> 
        <br/> 
        <input id="chatInput" type="text" style="width: 500px" value="" /> 
        <br/> 
        <button id="chatButton">Send</button> 
        <br/> 
    </div> 

    var outboundChatMessages = new 
     WebSocket('ws://localhost:8200/app/chatMessage.new'); 
    // Post new chat messages 
    outboundChatMessages.onopen = function(event) { 
      document.getElementById('chatButton') 
       .addEventListener('click', function () { 
         var chatInput = document.getElementById('chatInput'); 
         console.log('Publishing "' + chatInput.value + '"'); 
         outboundChatMessages.send(chatInput.value); 
         chatInput.value = ''; 
         chatInput.focus(); 
       }); 
    } 

    var inboundChatMessages = 
      new WebSocket('ws://localhost:8200/topic/chatMessage.new'); 
    // Listen for new chat messages 
    inboundChatMessages.onmessage = function (event) { 
      console.log('Received ' + event.data); 
      var chatDisplay = document.getElementById('chatDisplay');
      chatDisplay.value = chatDisplay.value + event.data + 'n'; 
    };

    @Bean 
    HandlerMapping webSocketMapping(CommentService commentService, 
      InboundChatService inboundChatService, 
      OutboundChatService outboundChatService) { 
        Map<String, WebSocketHandler> urlMap = new HashMap<>(); 
        urlMap.put("/topic/comments.new", commentService); 
        urlMap.put("/app/chatMessage.new", inboundChatService); 
        urlMap.put("/topic/chatMessage.new", outboundChatService); 
 
        Map<String, CorsConfiguration> corsConfigurationMap = 
         new HashMap<>(); 
        CorsConfiguration corsConfiguration = new CorsConfiguration(); 
        corsConfiguration.addAllowedOrigin("http://localhost:8080"); 
        corsConfigurationMap.put( 
          "/topic/comments.new", corsConfiguration); 
        corsConfigurationMap.put( 
          "/app/chatMessage.new", corsConfiguration); 
        corsConfigurationMap.put( 
          "/topic/chatMessage.new", corsConfiguration); 
 
        SimpleUrlHandlerMapping mapping = new 
         SimpleUrlHandlerMapping(); 
        mapping.setOrder(10); 
        mapping.setUrlMap(urlMap); 
        mapping.setCorsConfigurations(corsConfigurationMap); 
 
        return mapping; 
    } 

    public interface ChatServiceStreams { 
 
      String NEW_COMMENTS = "newComments"; 
      String CLIENT_TO_BROKER = "clientToBroker"; 
      String BROKER_TO_CLIENT = "brokerToClient"; 
 
      @Input(NEW_COMMENTS) 
      SubscribableChannel newComments(); 
 
      @Output(CLIENT_TO_BROKER) 
      MessageChannel clientToBroker(); 
 
      @Input(BROKER_TO_CLIENT) 
      SubscribableChannel brokerToClient(); 
    } 

    @Service 
    @EnableBinding(ChatServiceStreams.class) 
    public class InboundChatService implements WebSocketHandler { 
 
      private final ChatServiceStreams chatServiceStreams; 
 
      public InboundChatService(ChatServiceStreams chatServiceStreams)
      { 
        this.chatServiceStreams = chatServiceStreams; 
      } 
 
      @Override 
      public Mono<Void> handle(WebSocketSession session) { 
        return session 
         .receive() 
         .log("inbound-incoming-chat-message") 
         .map(WebSocketMessage::getPayloadAsText) 
         .log("inbound-convert-to-text") 
         .map(s -> session.getId() + ": " + s) 
         .log("inbound-mark-with-session-id") 
         .flatMap(this::broadcast) 
         .log("inbound-broadcast-to-broker") 
         .then(); 
      } 
 
      public Mono<?> broadcast(String message) { 
        return Mono.fromRunnable(() -> { 
          chatServiceStreams.clientToBroker().send( 
            MessageBuilder 
             .withPayload(message) 
             .build()); 
        }); 
      } 
    } 

    spring: 
      cloud: 
        stream: 
          bindings: 
            clientToBroker: 
              destination: learning-spring-boot-chat-user-messages 
              group: app-chatMessages 

    @Service 
    @EnableBinding(ChatServiceStreams.class) 
    public class OutboundChatService implements WebSocketHandler { 
 
      private final static Logger log = 
        LoggerFactory.getLogger(CommentService.class); 
 
      private Flux<String> flux; 
      private FluxSink<String> chatMessageSink; 
 
      public OutboundChatService() { 
        this.flux = Flux.<String>create( 
          emitter -> this.chatMessageSink = emitter, 
           FluxSink.OverflowStrategy.IGNORE) 
           .publish() 
           .autoConnect(); 
      } 
 
      @StreamListener(ChatServiceStreams.BROKER_TO_CLIENT) 
      public void listen(String message) { 
        if (chatMessageSink != null) { 
          log.info("Publishing " + message + 
            " to websocket..."); 
            chatMessageSink.next(message); 
        } 
      } 
 
      @Override 
      public Mono<Void> handle(WebSocketSession session) { 
        return session 
         .send(this.flux 
         .map(session::textMessage) 
         .log("outbound-wrap-as-websocket-message")) 
         .log("outbound-publish-to-websocket"); 
 
      } 
    }

    spring: 
      cloud: 
        stream: 
          bindings: 
            brokerToClient: 
              destination: learning-spring-boot-chat-user-messages 
              group: topic-chatMessages