1

我尝试使用 Angular、Spring RSocket、Spring Data Mongo Reactive 创建客户端/服务器示例。

完整的代码，请在此处查看。

后端代码：

@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())

在前端代码中。

@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 = '';
  }
}

我尝试使用requestStream方法连接消息路由，但失败了，在rscoket-jsAPI 中，似乎我必须在那里设置一个Subscription选项并设置请求号，否则它根本没有连接。

当我运行前端应用程序时，在浏览器控制台中出现以下错误。

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.

更新：阅读相关文档后，我自己解决了这个上限集合。默认情况下，Spring Data Mongo 不会为您创建上限集合。

现在我在客户端遇到另一个错误，有一个意外关闭错误导致RSocketWebsocketClient。

更新 2：已解决。data应该是当null服务器端不接受有效负载时。

最终的工作版本被合并到master中。

4

0 回答 0