Spring WebFluxをバックエンドにReactiveMongoRepositoryを使用し、Angular 4をフロントに使用して、アプリケーションの簡単なチャットモジュールに取り組んでいます。WebSocketSessionを介してデータを受信できますが、dbからすべてのメッセージをストリーミングした後、接続を維持してメッセージリストを更新します。誰かが私にそれを達成する方法の手掛かりを与えることができますか、あるいは私は間違った仮定に従っているのでしょうか?
WebSocketを担当するJavaバックエンド、私のサブスクライバーは現在の状態のみをログに記録し、そこには何も関係ありません。
WebFluxConfiguration:
@Configuration
@EnableWebFlux
public class WebSocketConfig {
private final WebSocketHandler webSocketHandler;
@Autowired
public WebSocketConfig(WebSocketHandler webSocketHandler) {
this.webSocketHandler = webSocketHandler;
}
@Bean
@Primary
public HandlerMapping webSocketMapping() {
Map<String, Object> map = new HashMap<>();
map.put("/websocket-messages", webSocketHandler);
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setOrder(10);
mapping.setUrlMap(map);
return mapping;
}
@Bean
public WebSocketHandlerAdapter handlerAdapter() {
return new WebSocketHandlerAdapter();
}
}
WebSocketHandlerの実装
@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<Void> 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);
}
}
}
とMongoRepo
@Repository
public interface MessageRepository extends
ReactiveMongoRepository<Message,String> {
}
フロントエンドの処理:
@Injectable()
export class WebSocketService {
private subject: Rx.Subject<MessageEvent>;
constructor() {
}
public connect(url): Rx.Subject<MessageEvent> {
if (!this.subject) {
this.subject = this.create(url);
console.log('Successfully connected: ' + url);
}
return this.subject;
}
private create(url): Rx.Subject<MessageEvent> {
const ws = new WebSocket(url);
const observable = Rx.Observable.create(
(obs: Rx.Observer<MessageEvent>) => {
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) => {
if (ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify(data));
}
}
};
return Rx.Subject.create(observer, observable);
}
}
他のサービスでは、私のタイプへの応答から観察可能にマッピングしています
constructor(private wsService: WebSocketService) {
this.messages = <Subject<MessageEntity>>this.wsService
.connect('ws://localhost:8081/websocket-messages')
.map((response: MessageEvent): MessageEntity => {
const data = JSON.parse(response.data);
return new MessageEntity(data.id, data.user_id, data.username, data.message, data.links);
});
}
そして最後に、接続が閉じられているために使用できないsend関数を使用したサブスクリプション:
ngOnInit() {
this.messages = [];
this._ws_subscription = this.chatService.messages.subscribe(
(message: MessageEntity) => {
this.messages.push(message);
},
error2 => {
console.log(error2.json());
},
() => {
console.log('Closed');
}
);
}
sendTestMessage() {
this.chatService.messages.next(new MessageEntity(null, '59ca30ac87e77d0f38237739', 'mickl', 'test message angular', null));
}
受信中のチャットメッセージがデータストアに永続化されていると仮定すると、Spring Data MongoDB Reactiveのカスタマイズ可能なカーソル機能を使用できます(リファレンスドキュメントを参照)。
したがって、次のようにリポジトリに新しいメソッドを作成できます。
public interface MessageRepository extends ReactiveSortingRepository< Message, String> {
@Tailable
Flux<Message> findWithTailableCursor();
}
テール可能カーソルにはいくつかの制限があることに注意してください。mongoコレクションにはキャップを付ける必要があり、エントリは挿入順にストリーミングされます。
Spring WebFlux WebSocketサポートはまだSTOMPもメッセージブローカーもサポートしていませんが、そのようなユースケースにはこれがより良い選択かもしれません。
この記事はインターネットから収集されたものであり、転載の際にはソースを示してください。
侵害の場合は、連絡してください[email protected]
コメントを追加