Come posso iscrivermi agli eventi SSE in entrata utilizzando Spring

Question

Ho scritto una primavera RestController che restituisce un SseEmitter (per evento inviato dal server) e aggiunge i collegamenti HATEOAS a ciascun evento. Ecco un esempio semplificato ma funzionamento di questo controllore:

package hello; 

import static org.springframework.hateoas.mvc.ControllerLinkBuilder.linkTo; 
import static org.springframework.hateoas.mvc.ControllerLinkBuilder.methodOn; 
import hello.Greeting.Status; 
import java.io.IOException; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.web.bind.annotation.RequestMapping; 
import org.springframework.web.bind.annotation.RequestParam; 
import org.springframework.web.bind.annotation.RestController; 
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; 
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; 

@RestController 
public class GreetingController { 

    private static final Logger log = LoggerFactory.getLogger(GreetingController.class); 

    private static final String template = "Hello, %s!"; 

    class GreetingRequestHandler implements Runnable { 

     private ResponseBodyEmitter emitter; 
     private Greeting greeting; 

     public GreetingRequestHandler(final ResponseBodyEmitter emitter, final Greeting greeting) { 
      this.emitter = emitter; 
      this.greeting = greeting; 
     } 

     @Override 
     public void run() { 
      try { 
       log.info(this.greeting.toString()); 
       this.emitter.send(this.greeting); 
       Thread.sleep(5000); 
       if (Status.COMPLETE.equals(this.greeting.getStatus())) { 
        this.emitter.complete(); 
       } else { 
        this.greeting.incrementStatus(); 
        new Thread(new GreetingRequestHandler(this.emitter, this.greeting)).start(); 
       } 
      } catch (IOException | InterruptedException e) { 
       e.printStackTrace(); 
      } 
     } 
    } 

    @RequestMapping(path = "/greeting") 
    public SseEmitter greeting(@RequestParam(value = "name", defaultValue = "World") final String name) { 
     SseEmitter emitter = new SseEmitter(); 
     Greeting greeting = new Greeting(String.format(template, name)); 
     greeting.add(linkTo(methodOn(GreetingController.class).greeting(name)).withSelfRel()); 
     new Thread(new GreetingRequestHandler(emitter, greeting)).start(); 
     log.info("returning emitter"); 
     return emitter; 
    } 
} 

La classe Greeting è la seguente:

package hello; 

import java.util.concurrent.atomic.AtomicInteger; 
import org.springframework.hateoas.ResourceSupport; 
import com.fasterxml.jackson.annotation.JsonCreator; 
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; 
import com.fasterxml.jackson.annotation.JsonProperty; 

@JsonIgnoreProperties(ignoreUnknown = true) 
public class Greeting extends ResourceSupport { 

    private final String content; 
    private final static AtomicInteger idProvider = new AtomicInteger(); 
    private int greetingId; 
    private Status status; 

    enum Status { 
     ENQUEUED, 
     PROCESSING, 
     COMPLETE; 
    } 

    @JsonCreator 
    public Greeting(@JsonProperty("content") final String content) { 
     this.greetingId = idProvider.addAndGet(1); 
     this.status = Status.ENQUEUED; 
     this.content = content; 
    } 

    public Status getStatus() { 
     return this.status; 
    } 

    protected void setStatus(final Status status) { 
     this.status = status; 
    } 

    public int getGreetingId() { 
     return this.greetingId; 
    } 

    public String getContent() { 
     return this.content; 
    } 

    @Override 
    public String toString() { 
     return "Greeting{id='" + this.greetingId + "', status='" + this.status + "' content='" + this.content + "', " + super.toString() + "}"; 
    } 

    public void incrementStatus() { 
     switch (this.status) { 
      case ENQUEUED: 
       this.status = Status.PROCESSING; 
       break; 
      case PROCESSING: 
       this.status = Status.COMPLETE; 
       break; 
      default: 
       break; 
     } 
    } 
} 

Questo codice funziona perfettamente. Se provo a raggiungere il servizio REST utilizzando un browser Web, vedo eventi che appaiono con contenuto e link corretti.

Il risultato sembra (ogni evento appare 5 secondi dopo il precedente):

data:{"content":"Hello, Kraal!","greetingId":8,"status":"ENQUEUED","_links":{"self":{"href":"http://localhost:8080/greeting?name=Kraal"}}} 
data:{"content":"Hello, Kraal!","greetingId":8,"status":"PROCESSING","_links":{"self":{"href":"http://localhost:8080/greeting?name=Kraal"}}} 
data:{"content":"Hello, Kraal!","greetingId":8,"status":"COMPLETE","_links":{"self":{"href":"http://localhost:8080/greeting?name=Kraal"}}} 

Ora ho bisogno di chiamare questo servizio REST e leggere questi eventi da un'altra applicazione primavera ... Ma non ho indovina come scrivere il codice cliente usando Spring. Questo non funziona come RestTemplate è stato progettato per il client di accesso laterale sincrona HTTP ...

ObjectMapper mapper = new ObjectMapper(); 
    mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); 
    mapper.registerModule(new Jackson2HalModule()); 

    // required for HATEOAS 
    MappingJackson2HttpMessageConverter converter = new MappingJackson2HttpMessageConverter(); 
    converter.setSupportedMediaTypes(MediaType.parseMediaTypes("application/hal+json")); 
    converter.setObjectMapper(mapper); 

    // required in order to be able to read serialized objects 
    MappingJackson2HttpMessageConverter converter2 = new MappingJackson2HttpMessageConverter(); 
    converter2.setSupportedMediaTypes(MediaType.parseMediaTypes("application/octet-stream")); 
    converter2.setObjectMapper(mapper); 

    // required to understand SSE events 
    MappingJackson2HttpMessageConverter converter3 = new MappingJackson2HttpMessageConverter(); 
    converter3.setSupportedMediaTypes(MediaType.parseMediaTypes("text/event-stream")); 

    List<HttpMessageConverter<?>> converters = new ArrayList<HttpMessageConverter<?>>(); 
    converters.add(converter); 
    converters.add(converter2); 
    converters.add(converter3); 

    // probably wrong template 
    RestTemplate restTemplate = new RestTemplate(); 
    restTemplate = new RestTemplate(converters); 
    // this does not work as I receive events and no a single object 
    Greeting greeting = restTemplate.getForObject("http://localhost:8080/greeting/?name=Kraal", Greeting.class); 
    log.info(greeting.toString()); 

Il messaggio di errore che ottiene è:

Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'data': was expecting ('true', 'false' or 'null') 

Infatti ogni evento è un evento di SSE e inizia con "i dati :" ...

Quindi le domande sono:

• quale modulo ObjectMapper dovrei registrare per poter mappare SSE con Jackson?
• Come posso iscrivermi agli eventi SSE in arrivo (pattern osservatore) utilizzando Spring?

Grazie in anticipo.

Nota laterale: Mentre sto facendo fatica usando Spring, ho provato a farlo utilizzando il supporto SSE di Jersey come segue. Utilizzando Jersey ricevo eventi come previsto, ma poi non riesco a lanciare ad una classe Greeting (per lo stesso motivo di cui sopra Credo che è che io non ho il modulo di conversione a destra.):

Client client = ClientBuilder.newBuilder().register(converter).register(SseFeature.class).build(); 
WebTarget target = client.target("http://localhost:8080/greeting/?name=Kraal"); 
EventInput eventInput = target.request().get(EventInput.class); 
while (!eventInput.isClosed()) { 
    final InboundEvent inboundEvent = eventInput.read(); 
    if (inboundEvent == null) { 
     // connection has been closed 
     break; 
    } 
    // this works fine and prints out events as they are incoming 
    System.out.println(inboundEvent.readData(String.class)); 
    // but this doesn't as no proper way to deserialize the 
    // class with HATEOAS links can be found 
    // Greeting greeting = inboundEvent.readData(Greeting.class); 
    // System.out.println(greeting.toString()); 
} 

Answer 1

come da documentation

è possibile utilizzare inboundEvent.readData(Class<T> type)