स्प्रिंग एसएसईमीटर, विधि भेजने के बाद ईवेंट नहीं भेजे जाते हैं

Question

मैं स्प्रिंग 4 (टॉमकैट 7, सर्वलेट-एपीआई 3.0.1) के साथ सर्वर से भेजे गए ईवेंट बनाने की कोशिश कर रहा हूं।

समस्या यह है कि मेरी Events विधि भेजने के बाद सही नहीं भेजा गया है। वे सभी के समय के बाद, केवल 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); 
}; 

अनुप्रयोग 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 परीक्षण भाग गया। मैंने जो कुछ भी ऑनलाइन पढ़ा है, उससे एसएसईईमीटर का उपयोग Reactive Streams, जैसे RxJava के कुछ कार्यान्वयन के साथ किया जाना है। यह थोड़ा जटिल है, लेकिन यह निश्चित रूप से काम करता है। विचार यह है कि आप एमिटर, और एक पर्यवेक्षक बनाते हैं, और उत्तरार्द्ध को प्रकाशक को सब्सक्राइब करते हैं। प्रकाशक अपने व्यवहार को एक अलग थ्रेड में निष्पादित करता है, आउटपुट तैयार होने पर अवलोकन को सूचित करता है, और अवलोकन करने योग्य 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 इस मॉडल के कुछ उदाहरण हैं, और आप पा सकते हैं Goxling 'RxJava SseEmitter' द्वारा अधिक। प्रतिक्रियाशील स्ट्रीम/RxJava/SseEmitter इंटरैक्शन को ग्रोक करने में कुछ समय लगता है, लेकिन एक बार ऐसा करने के बाद यह बहुत ही सुरुचिपूर्ण है। उम्मीद है कि यह आपको सही रास्ते पर सेट करता है!

Answer 2

जबकि अन्य जवाब सही है, अगर आप इसे अपने आप को प्रबंधित करने के लिए कॉल कर सकते हैं चाहते हैं:

emitter.complete()