发布不同对象类型的 FluxSink答案

【问题标题】：Publish FluxSink of different object types发布不同对象类型的 FluxSink
【发布时间】：2020-02-27 12:25:03
【问题描述】：

我有一个响应通量的 rsocket 端点：

@MessageMapping("responses")
Flux<?> deal(@Payload String message) {
    return myService.generateResponses(message);
}

响应可以是使用以下代码（如果有效）异步生成的 3 种不同类型的对象中的任何一种：

public Flux<?> generateResponses(String request) {
  // Setup response sinks
  final FluxProcessor publish = EmitterProcessor.create().serialize();
  final FluxSink<Response1> sink1 = publish.sink();    
  final FluxSink<Response2> sink2 = publish.sink();
  final FluxSink<Response3> sink3 = publish.sink();

  // Get async responses: starts new thread to gather responses and update sinks
  new MyResponses(request, sink1, sink2, sink3)

  // Return the Flux
  Flux<?> output = Flux
    .from(publish
    .log());
}

问题在于，当我用不同的对象填充接收器时，实际上只有第一个接收器会发布回订阅者。

public class MyResponses extends CacheListenerAdapter {
  private FluxSink<Response1> sink1;
  private FluxSink<Response2> sink2;
  private FluxSink<Response3> sink3;

  // Constructor is omitted for brevity

  @Override
  public void afterCreate(EntryEvent event) {       
    if (event.getNewValue() instanceof Response1) {  
        Response1 r1 = (Response1)event.getNewValue();
        sink1.next(r1);
    }
    if (event.getNewValue() instanceof Response2) {  
        Response2 r2 = (Response2)event.getNewValue();
        sink2.next(r2);
    }
    if (event.getNewValue() instanceof Response3) {  
        Response3 r3 = (Response3)event.getNewValue();
        sink3.next(r3);
    }
  }
}

如果我制作<?> 类型的接收器，则会出现.next 错误：

The method next(capture#2-of ?) in the type FluxSink<capture#2-of ?> is not applicable for the arguments (Response1)

有没有更好的方法来满足这个要求？

【问题讨论】：

标签： project-reactor rsocket

【解决方案1】：

这不适用于不同对象的原因是与底层对象类型的 Spring Boot Data Geode 序列化有关。让对象 Flux 工作的方法是使用 1 个 <Object> 类型的接收器

public Flux<Object> generateResponses(String request) {
  // Setup the Flux
  EmitterProcessor<Object> emitter = EmitterProcessor.create();
  FluxSink<Object> sink = emitter.sink(FluxSink.OverflowStrategy.LATEST);

  // Get async responses: starts new thread to gather responses and update sinks
  new MyResponses(request, sink)

  // Setup an output Flux to publish the input Flux
  Flux<Object> out = Flux
    .from(emitter
    .log(log.getName()));
}

然后事件处理程序使用了 1 个接收器

public class MyResponses extends CacheListenerAdapter {
  private FluxSink<Object> sink;

  // Constructor is omitted for brevity

  @Override
  public void afterCreate(EntryEvent event) {       
    if (event.getNewValue() instanceof Response1) {  
        Response1 r1 = (Response1)event.getNewValue();
        sink.next(r1);
    }
    if (event.getNewValue() instanceof Response2) {  
        Response2 r2 = (Response2)event.getNewValue();
        sink.next(r2);
    }
    if (event.getNewValue() instanceof Response3) {  
        Response3 r3 = (Response3)event.getNewValue();
        sink.next(r3);
    }
  }
}

【讨论】：