Rsocket JS issue when connecting Spring RSocket based backend

Question

I tried to create a client/server sample using Angular, Spring RSocket, Spring Data Mongo Reactive.

The complete codes, check here.

The backend code:

@SpringBootApplication
class RSocketServerApplication

fun main(args: Array<String>) {
    runApplication<RSocketServerApplication>(*args)
}

@Controller
class MessageController(private val messages: MessageRepository) {
    @MessageMapping("send")
    fun hello(p: String) = this.messages.save(ChatMessage(body = p, sentAt = Instant.now())).log().then()

    @MessageMapping("messages")
    fun messageStream(): Flux<ChatMessage> = this.messages.getMessagesBy().log()
}

interface MessageRepository : ReactiveMongoRepository<ChatMessage, String> {
    @Tailable
    fun getMessagesBy(): Flux<ChatMessage>
}

@Document
data class ChatMessage(@Id var id: String? = null, var body: String, var sentAt: Instant = Instant.now())

In the frontend code.

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit, OnDestroy {

  title = 'client';
  message = '';
  messages: string[];
  client: RSocketClient;
  sub = new Subject();

  ngOnInit(): void {
    this.messages = [];

    // Create an instance of a client
    this.client = new RSocketClient({
      serializers: {
        data: JsonSerializer,
        metadata: IdentitySerializer
      },
      setup: {
        // ms btw sending keepalive to server
        keepAlive: 60000,
        // ms timeout if no keepalive response
        lifetime: 180000,
        // format of `data`
        dataMimeType: 'application/json',
        // format of `metadata`
        metadataMimeType: 'message/x.rsocket.routing.v0',
      },
      transport: new RSocketWebSocketClient({
        url: 'ws://localhost:8080/rsocket'
      }),
    });

    // Open the connection
    this.client.connect().subscribe({
      onComplete: (socket: RSocket) => {

        // socket provides the rsocket interactions fire/forget, request/response,
        // request/stream, etc as well as methods to close the socket.
        socket
          .requestStream({
            data: "",
            metadata: String.fromCharCode('messages'.length) + 'messages'
          })
          .subscribe({
            onComplete: () => console.log('complete'),
            onError: error => {
              console.log(error);
              this.addErrorMessage("Connection has been closed due to ", error);
            },
            onNext: payload => {
              console.log(payload);
              this.addMessage(payload.data);
            },
            onSubscribe: subscription => {
              subscription.request(1);
            },
          });

        this.sub.subscribe({
          next: (data) => {
            socket.fireAndForget({
              data: data,
              metadata: String.fromCharCode('send'.length) + 'send',
            });
          }
        })
      },
      onError: error => {
        console.log(error);
        this.addErrorMessage("Connection has been refused due to ", error);
      },
      onSubscribe: cancel => {
        /* call cancel() to abort */
      }
    });
  }

  addErrorMessage(description: string, error: any) {
    console.log(description + ', ' + error);
  }

  addMessage(newMessage: string) {
    this.messages = [...this.messages, newMessage];
  }

  ngOnDestroy(): void {
    this.sub.unsubscribe();
    if (this.client) {
      this.client.close();
    }
  }

  sendMessage() {
    console.log("sending message:" + this.message);
    this.sub.next(this.message);
    this.message = '';
  }
}

I tried to use requestStream method to connect the messages route, but fialed, in the rscoket-js API, it seems I have to setup a Subscription option and set request number there, or it does not conncet at all.

When I ran the frontend application, I got the following error in the browser console.

Error: RSocket error 0x201 (APPLICATION_ERROR): Query failed with error code 2 and error message 'error processing query: ns=chat.chatMessage batchSize=32Tree: $and
Sort: {}
Proj: {}
 tailable cursor requested on non capped collection' on server localhost:27017; nested exception is com.mongodb.MongoQueryException: Query failed with error code 2 and error message 'error processing query: ns=chat.chatMessage batchSize=32Tree: $and
Sort: {}
Proj: {}
 tailable cursor requested on non capped collection' on server localhost:27017. See error `source` property for details.
    createErrorFromFrame RSocketFrame.js:189
    _handleStreamFrame RSocketMachine.js:625
    _handleFrame RSocketMachine.js:192
    onNext Flowable.js:233
    _handleMessage RSocketWebSocketClient.js:52
    _handleMessage RSocketWebSocketClient.js:52
    Angular 7
app.component.ts:58:22
Connection has been closed due to , Error: RSocket error 0x201 (APPLICATION_ERROR): Query failed with error code 2 and error message 'error processing query: ns=chat.chatMessage batchSize=32Tree: $and
Sort: {}
Proj: {}
 tailable cursor requested on non capped collection' on server localhost:27017; nested exception is com.mongodb.MongoQueryException: Query failed with error code 2 and error message 'error processing query: ns=chat.chatMessage batchSize=32Tree: $and
Sort: {}
Proj: {}
 tailable cursor requested on non capped collection' on server localhost:27017. See error `source` property for details.

Update: I have resolved this capped collections myself after reading the related docs. Spring Data Mongo does not create a capped collection for you by default.

Now I encountered another error in the client, there is a close unexpectedly error caused by RSocketWebsocketClient.

Update 2: Resolved. the data should be null when the server side does not accept payload.

The final working version is merged into master.