如何使用Spring订阅传入的SSE事件

Question

我编写了一个Spring RestController，它返回一个SseEmitter（对于服务器发送的事件），并向每个事件添加HATEOAS链接。以下是该控制器的一个简化但有效的示例：

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;
    }
}

问候语课程如下：

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;
        }
    }
}

此代码运行完美。如果我尝试使用Web浏览器访问REST服务，我会看到显示正确内容和链接的事件。

结果如下所示（每个事件出现在前一个事件之后5秒）：

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"}}}

现在我需要调用这个REST服务并从另一个Spring应用程序读取这些事件。。。但我不知道如何使用Spring编写客户机代码。这不起作用，因为RestTemplate是为同步客户端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());

我得到的错误消息是：

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

事实上，每个事件都是一个SSE事件，并以“data：”开头...

因此，问题是：

• 我应该注册什么ObjectMapper模块才能与Jackson一起映射SSE？
• 如何使用Spring订阅传入的SSE事件（观察者模式）？

提前谢谢。

旁注：当我使用Spring努力做到这一点时，我尝试使用泽西SSE支持来做到这一点，如下所示。使用泽西我按预期接收事件，但我无法将它们强制转换为问候类（出于与上述相同的原因，我想这是我没有正确的转换器模块。）：

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

根据文件

您可以使用inboundEvent。readData（类