0

Webシステムにおいて、同じ画面を複数ユーザーが表示・操作している状態で画面を同期化したいと考えています。

○期待している動作
ユーザが画面を操作>AjaxでDBを更新>更新内容を、同じ画面を開いている別ユーザの画面にも反映

○現状
play.libs.EventSourceのリファレンスに乗っているサンプルコードを実装し、一定時間ごとにサーバからクライアントへデータを送信できることを確認しました。

・javascript

$(function() {
/**************************************************************************
 *
 * 同期処理用SSE
 *
 **************************************************************************/
var eventSource = new EventSource("/xxx/sse/test");
eventSource.onmessage = function (event) {
      alert(event.data);
    };
});  

・サーバ側

public Result testSSE() {
    Source<String, ?> tickSource = Source.tick(Duration.Zero(), Duration.create(10000, TimeUnit.MILLISECONDS), "TICK");
    Source<EventSource.Event, ?> eventSource = tickSource.map((tick) -> EventSource.Event.event(df.format(ZonedDateTime.now())));
    return ok().chunked(eventSource.via(EventSource.flow())).as(Http.MimeTypes.EVENT_STREAM);
}

○疑問
このサンプルを参考に、「一定時間ごと」ではなく「DB更新処理が完了する毎」にフラグを変更したいのですが、方法がわからずに頓挫している状態です。
ご存知の方がいらっしゃいましたらご協力をお願いします・・・。

○仮説
AkkaStreamで時間ごとに繰り返してるようなのでその処理を削り、DB保存処理の完了後にeventSourceを作成してやればいくのでは?と考えました。
しかし結果として、DB保存処理を走らせていないにも関わらず、3秒程度の間隔でクライアント側にメッセージが表示されました。
Source.singleの意味合いを勘違いしていると思うのですが、現状はここで行き詰っています。

・javascript

$(function() {
/**************************************************************************
 *
 * 同期処理用SSE
 *
 **************************************************************************/
var eventSource = new EventSource("/xxx/sse/test2");
eventSource.onmessage = function (event) {
      alert(event.data);
    };
});

・サーバ側

public Source<EventSource.Event, ?> eventSource = Source.single(EventSource.Event.event("test Datas"));    
public Result saveDB() {
    ...
    eventSource = Source.single(EventSource.Event.event("test Datas"));
}
public Result testSSE2() {
    return ok().chunked(eventSource.via(EventSource.flow())).as(Http.MimeTypes.EVENT_STREAM);
}

1 件の回答 1

1

しかし結果として、DB保存処理を走らせていないにも関わらず、3秒程度の間隔でクライアント側にメッセージが表示されました。

3秒おきに発生する理由は、ブラウザのEventSourceのreconnection timeの初期値が3秒で、3秒ごとに再接続しているからだと思います。再接続間隔はサーバーから送信するメッセージのretryフィールドで設定可能です。

https://html.spec.whatwg.org/multipage/comms.html#the-eventsource-interface

A reconnection time, in milliseconds. This must initially be a user-agent-defined value, probably in the region of a few seconds.

Server-Sent Events の利用 - Server-sent events | MDN
https://developer.mozilla.org/ja/docs/Server-sent_events/Using_server-sent_events#Fields

EventSource 覚え書き - JavaScriptで遊ぶよ - g:javascript
http://javascript.g.hatena.ne.jp/edvakf/20101011/1286853847


本題ですが、期待している動作を実現するためには下記のような実装を行う必要があるかと思います。

  1. DB更新リクエストでDBを更新
  2. 更新された内容をqueueing
  3. DB更新リクエストのレスポンスを返す
  4. queueingされた内容をSSEで送信

queueingしてSSEで送信する方法はいくつかありそうです。

scala - How to create a Source that can receive elements later via a method call? - Stack Overflow
https://stackoverflow.com/a/33415214/4366193

例えばActor(Source.actorRef())を使用して簡易的に実装すると下記のようになるかと思います。
Source.actorRef()のActorRefを保存し、そのactorへDBに保存したデータを送ります。

@Singleton
public class EventSourceController extends Controller {

    private final ActorRef messageActor;

    @Inject
    public EventSourceController(final ActorSystem system) {
        messageActor = system.actorOf(Props.create(MessageActor.class));
    }

    public Result saveDB() {

        // DBへ値が正常に保存されたら、`messageActor`経由で
        // DBに保存されたデータ(Message)を送信
        messageActor.tell(new Message("new data"), ActorRef.noSender());

        return ok();
    }

    public Result stream() {
        final Source<Message, ?> source = Source.<Message>actorRef(256, OverflowStrategy.dropHead())
                .mapMaterializedValue(sourceActor -> {
                    messageActor.tell(sourceActor, ActorRef.noSender());
                    return null;
                });
        return ok()
                .chunked(source.map(msg -> EventSource.Event.event(msg.getData())).via(EventSource.flow()))
                .as(Http.MimeTypes.EVENT_STREAM);
    }

}

class Message {
    private final String data;

    public Message(final String data) {
        this.data = data;
    }

    public String getData() {
        return data;
    }
}

class MessageActor extends AbstractActor {

    private final List<ActorRef> actors = new ArrayList<>();

    public MessageActor() {
        receive(ReceiveBuilder.
                match( ActorRef.class, actorRef -> {
                    // actorの停止をwatchする
                    context().watch(actorRef);
                    actors.add(actorRef);
                }).
                match( Message.class, msg -> {
                    // 保存していたactorへ DBに保存したデータ(Message)を送信する
                    actors.forEach(actor -> actor.tell(msg, self()));
                }).
                match( Terminated.class, terminated -> {
                    // 停止したactorをリストから削除する
                    actors.remove(terminated.actor());
                }).build()
        );
    }
}

必要最小限のコードにするため、ActorRefを直接メッセージとして送信するなど不適切な箇所がいくつかありますので、適切に実装しなおしてください。


Source.actorPublisher()を使用した場合の例では下記のページが参考になるかと思います。

Developer Without Future: Play and SSE (with Akka Streams)
http://developerwithoutfuture.blogspot.jp/2017/01/play-and-sse-with-akka-streams.html

https://github.com/darkwings/play-sse

1
  • ご挨拶が遅れてしまい申し訳ありません。 ブラウザ側の仕様や、基本的なEventSourseの資料、サンプルコードまでくださいまして有難うございます! なんとか動作させることができました! Commented 2017年2月15日 4:47

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。