The last step in implementing user-to-user messages is to apply a filter to OutboundChatService. Since we coded up UserParsingHandshakeHandler, we have to adjust the service to handle this:

    @Service 
    @EnableBinding(ChatServiceStreams.class) 
    public class OutboundChatService 
     extends UserParsingHandshakeHandler { 
       ... 
    } 

For starters, we need to change this class to extend UserParsingHandshakeHandler instead of WebSocketHandler.

There's no need to alter the constructor call where our FluxSink is configured. However, the handler itself must be adjusted as follows:

    @Override 
    protected Mono<Void> handleInternal(WebSocketSession session) { 
      return session 
       .send(this.flux 
          .filter(s -> validate(s, getUser(session.getId()))) 
          .map(this::transform) 
          .map(session::textMessage) 
          .log(getUser(session.getId()) + 
              "-outbound-wrap-as-websocket-message")) 
      .log(getUser(session.getId()) + 
          "-outbound-publish-to-websocket"); 
    } 

The details can be explained as follows:

• Just like InboundChatService, we must now implement handleInternal(WebSocketSession).
• It has the same session.send(Flux) call, but that Flux has a couple of extra steps added, including a filter and an extra map.
• The filter call validates each message, deciding whether or not this user should get it. (We'll write that validate() method in a moment).
• Assuming the message is valid for this user, it uses a local transform method to tweak it.
• The rest of the machinery used to convert this string message into a WebSocketMessage<String> and pipe it over the WebSocket is the same as before.

When dealing with streams of messages, layering in a filter is no biggie. See how in the following code:

    private boolean validate(Message<String> message, String user) { 
      if (message.getPayload().startsWith("@")) { 
        String targetUser = message.getPayload() 
            .substring(1, message.getPayload().indexOf(" ")); 
 
        String sender = message.getHeaders() 
            .get(ChatServiceStreams.USER_HEADER, String.class); 
 
        return user.equals(sender) || user.equals(targetUser); 
      } else { 
        return true; 
      } 
    }

This last code can be described as follows:

• validate accepts a Message<String> and the name of the current user (not the user that sent the message).
• It first checks the payload, and if it starts with @, it looks deeper. If the message does NOT start with @, it just lets it on through.
• If the message starts with @, it proceeds to extract the target user by parsing the text between @ and the first space. It also extracts the original sender of the message using the User header.
• If the current user is either the sender or the receiver, the message is allowed through. Otherwise, it is dropped.

A filtering function like this makes it easy to layer various options. We used it to target user-specific messages. But imagine putting things like security checks, regional messages, time-based messages, and more!

To wrap this up, we need to also code a little transformation to make the user-to-user experience top notch:

    private String transform(Message<String> message) { 
      String user = message.getHeaders() 
        .get(ChatServiceStreams.USER_HEADER, String.class); 
      if (message.getPayload().startsWith("@")) { 
        return "(" + user + "): " + message.getPayload(); 
      } else { 
          return "(" + user + ")(all): " + message.getPayload(); 
      } 
    } 

This preceding nice little transformation can be described as follows:

• transform accepts a Message<String>, and converts it into a plain old string message
• It extracts the User header to find who wrote the message
• If the message starts with @, then it assumes the message is targeted, and prefixes it with the author wrapped in parentheses
• If the message does NOT start with @, then it prefixes it with the author wrapped in parentheses plus (all), to make it clear that this is a broadcast message

With this change in place, we have coded a sophisticated user-to-user chat service, running on top of RabbitMQ, using Reactive Streams.