title: Spring WebFlux reactive WebSocket prevent connection closing
body: <p>I'm working on simple chat module for my application using Spring WebFlux with ReactiveMongoRepository on backend and Angular 4 on front. I'm able to receive data through WebSocketSession but after streaming all messages from db i want to keep the connection so i could update message list. Can anyone give me clues how to achieve that, or maybe i'm following wrong assumptions ?</p>
  
  <p>Java Backend responsible for WebSocket, my subscriber only logs current state, nothing relevant there:</p>
  
  <p>WebFluxConfiguration:</p>
  
  <pre><code>@Configuration
  @EnableWebFlux
  public class WebSocketConfig {
  
  private final WebSocketHandler webSocketHandler;
  
  @Autowired
  public WebSocketConfig(WebSocketHandler webSocketHandler) {
      this.webSocketHandler = webSocketHandler;
  }
  
  @Bean
  @Primary
  public HandlerMapping webSocketMapping() {
      Map&lt;String, Object&gt; map = new HashMap&lt;&gt;();
      map.put("/websocket-messages", webSocketHandler);
  
      SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
      mapping.setOrder(10);
      mapping.setUrlMap(map);
      return mapping;
  }
  
  @Bean
  public WebSocketHandlerAdapter handlerAdapter() {
      return new WebSocketHandlerAdapter();
  }
  
  
  }
  </code></pre>
  
  <p>WebSocketHandler Implementation</p>
  
  <pre><code>@Component
  public class MessageWebSocketHandler implements WebSocketHandler {
  
  private final MessageRepository messageRepository;
  private ObjectMapper mapper = new ObjectMapper();
  private MessageSubscriber subscriber = new MessageSubscriber();
  
  @Autowired
  public MessageWebSocketHandler(MessageRepository messageRepository) {
      this.messageRepository = messageRepository;
  }
  
  @Override
      public Mono&lt;Void&gt; handle(WebSocketSession session) {
      session.receive()
              .map(WebSocketMessage::getPayloadAsText)
              .map(this::toMessage)
              .subscribe(subscriber::onNext, subscriber:: onError, subscriber::onComplete);
      return session.send(
              messageRepository.findAll()
                      .map(this::toJSON)
                      .map(session::textMessage));
  }
  
  private String toJSON(Message message) {
      try {
          return mapper.writeValueAsString(message);
      } catch (JsonProcessingException e) {
          throw new RuntimeException(e);
      }
  }
  
  private Message toMessage(String json) {
      try {
          return mapper.readValue(json, Message.class);
      } catch (IOException e) {
          throw new RuntimeException("Invalid JSON:" + json, e);
      }
  }
  }
  </code></pre>
  
  <p>and MongoRepo</p>
  
  <pre><code>@Repository
  public interface MessageRepository extends 
  ReactiveMongoRepository&lt;Message,String&gt; {
  }
  </code></pre>
  
  <p>FrontEnd Handling:</p>
  
  <pre><code>@Injectable()
  export class WebSocketService {
    private subject: Rx.Subject&lt;MessageEvent&gt;;
  
    constructor() {
    }
  
    public connect(url): Rx.Subject&lt;MessageEvent&gt; {
      if (!this.subject) {
        this.subject = this.create(url);
        console.log('Successfully connected: ' + url);
      }
      return this.subject;
    }
  
    private create(url): Rx.Subject&lt;MessageEvent&gt; {
      const ws = new WebSocket(url);
      const observable = Rx.Observable.create(
        (obs: Rx.Observer&lt;MessageEvent&gt;) =&gt; {
          ws.onmessage = obs.next.bind(obs);
          ws.onerror = obs.error.bind(obs);
          ws.onclose = obs.complete.bind(obs);
          return ws.close.bind(ws);
        });
      const observer = {
        next: (data: Object) =&gt; {
          if (ws.readyState === WebSocket.OPEN) {
            ws.send(JSON.stringify(data));
          }
        }
      };
      return Rx.Subject.create(observer, observable);
    }
  }
  </code></pre>
  
  <p>in other service i'm mapping observable from response to my type</p>
  
  <pre><code>  constructor(private wsService: WebSocketService) {
      this.messages = &lt;Subject&lt;MessageEntity&gt;&gt;this.wsService
        .connect('ws://localhost:8081/websocket-messages')
        .map((response: MessageEvent): MessageEntity =&gt; {
          const data = JSON.parse(response.data);
          return new MessageEntity(data.id, data.user_id, data.username, data.message, data.links);
        });
    }
  </code></pre>
  
  <p>and finally subscribtion with send function which i can't use because of closed connection:</p>
  
  <pre><code>  ngOnInit() {
      this.messages = [];
      this._ws_subscription = this.chatService.messages.subscribe(
        (message: MessageEntity) =&gt; {
          this.messages.push(message);
        },
        error2 =&gt; {
          console.log(error2.json());
        },
        () =&gt; {
          console.log('Closed');
        }
      );
    }
  
    sendTestMessage() {
      this.chatService.messages.next(new MessageEntity(null, '59ca30ac87e77d0f38237739', 'mickl', 'test message angular', null));
    }
  </code></pre>