Spring sseEmitter、メソッドの送信直後にイベントが送信されない

Question

Spring 4（tomcat 7、servlet-api 3.0.1）でサーバー送信イベントを作成しようとしています。

私のEventsの問題は、メソッドの送信直後に送信されません。これらはすべて、タイムアウトがSseEmitterで、EventSourceのエラーイベントが発生した後にのみ、（タイムスタンプを使用して）同時にクライアントに送信されます。そして、クライアントは再接続しようとしています。何が起きているのか？

sse = new EventSource(urlBuilder(base, url)); 
sse.addEventListener('ping', function (event) { 
    dfd.notify(event); 
}); 

sse.addEventListener('message', function(event){ 
    dfd.notify(event); 
}); 

sse.addEventListener('close', function(event){ 
    dfd.notify(event); 
}); 

sse.onerror = function (error) { 
    console.log(error); 
}; 

sse.onmessage = function (event){ 
    dfd.notify(event); 
}; 

のApp initalizerコード

public class WebAppInitializer implements WebApplicationInitializer { 
    @Override 
    public void onStartup(ServletContext servletContext) throws ServletException { 
     AnnotationConfigWebApplicationContext ctx = new AnnotationConfigWebApplicationContext(); 
     ctx.register(AppConfig.class); 
     ctx.setServletContext(servletContext); 
     ctx.refresh(); 

     ServletRegistration.Dynamic dynamic = servletContext.addServlet("dispatcher", new DispatcherServlet(ctx)); 
     dynamic.setAsyncSupported(true); 
     dynamic.addMapping("/api/*"); 
     dynamic.setLoadOnStartup(1); 
     dynamic.setMultipartConfig(ctx.getBean(MultipartConfigElement.class)); 

     javax.servlet.FilterRegistration.Dynamic filter = servletContext 
       .addFilter("StatelessAuthenticationFilter", 
         ctx.getBean("statelessAuthenticationFilter", StatelessAuthenticationFilter.class)); 
     filter.setAsyncSupported(true); 
     filter.addMappingForUrlPatterns(null, false, "/api/*"); 

     filter = servletContext.addFilter("HibernateSessionRequestFilter", 
       ctx.getBean("hibernateSessionRequestFilter", HibernateSessionRequestFilter.class)); 
     filter.setAsyncSupported(true); 
     filter.addMappingForUrlPatterns(null, false, "/api/user/*"); 
    } 
} 

AppConfig.java

@Configuration 
@ComponentScan("ru.esoft.workflow") 
@EnableWebMvc 
@PropertySource({"classpath:mail.properties", "classpath:fatclient.properties"}) 
@EnableAsync 
@EnableScheduling 
public class AppConfig extends WebMvcConfigurerAdapter { 
... 
} 

0：クライアントコードで

@RequestMapping(value = "subscribe", method = RequestMethod.GET) 
public SseEmitter subscribe() throws IOException { 
    final SseEmitter emitter = new SseEmitter(); 
    Executors.newScheduledThreadPool(1).scheduleAtFixedRate(new Runnable() { 
     @Override 
     public void run() { 
      try { 
       emitter.send(SseEmitter.event().data("Thread writing: " + Thread.currentThread()).name("ping")); 
      } catch (Exception e) { 
      } 
     } 
    } , 1000, 1000, TimeUnit.MILLISECONDS); 
    return emitter; 
} 

：

私は単純なサービスを作成しました私のクライアントログの

画像：

Answer 1

SSEEmittersをテストするときに、私はこれに自分自身を走りました。私がオンラインで読んだことのすべてから、SSEEmittersは、RxJavaのように、Reactive Streamsの実装と組み合わせて使用​​することを意図しています。少し複雑ですが、間違いなく機能します。アイデアは、エミッタとObservableを作成し、後者をパブリッシャにサブスクライブするというものです。パブリッシャーはその動作を別のスレッドで実行し、出力が準備完了であることをObservableに通知し、observableがemitter.sendをトリガーします。ここであなたが望む何をすべき例の抜粋である：ここで

@RequestMapping("/whatever") 
public SseEmitter index( 
    SseEmitter emitter = new SseEmitter(); 
    Publisher<String> responsePublisher = someResponseGenerator.getPublisher(); 
    Observable<String> responseObservable = RxReactiveStreams.toObservable(responsePublisher); 

    responseObservable.subscribe(
     str -> { 
      try { 
       emitter.send(str); 
      } catch (IOException ex) { 
       emitter.completeWithError(ex); 
      } 
     }, 
     error -> { 
      emitter.completeWithError(error); 
     }, 
     emitter::complete 
     ); 

     return emitter; 
}; 

は、対応する出版社です：

public class SomeResponseGenerator {  
    public Publisher<String> getPublisher() { 
     Publisher<String> pub = new Publisher<String>() { 
      @Override 
      public void subscribe(Subscriber subscriber) { 
       Executors.newScheduledThreadPool(1).scheduleAtFixedRate(new Runnable() { 
        @Override 
        public void run() { 
         subscriber.onNext("Thread writing: " + Thread.currentThread().getName()); 
        } 
       }, 1000, 1000, TimeUnit.MILLISECONDS); 
      } 
     }; 

     return pub; 
    } 
} 

あり、このモデルのいくつかの例オンラインhereとhereがあり、そしてあなたが見つけることができますGoogling 'RxJava SseEmitter'のその他の記事Reactive Streams/RxJava/SseEmitterのやりとりを実行するには時間がかかりますが、いったんやり直すとかなりエレガントです。これがあなたの正しい道を歩むことを願っています！

Answer 2

他の答えは、あなたがそれを自分で管理したい場合は、あなたが呼び出すことができ、正しいですが：

emitter.complete()