代码之家 › 专栏 › 技术社区 › rupweb

发布不同对象类型的FluxSink

rsocket project-reactor

0

rupweb · 技术社区 · 6 年前

我有一个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)

有没有更好的方法来满足这一要求?

0 回复 | 直到 6 年前

1

0

rupweb 6 年前

这不适用于不同对象的原因与底层对象类型的Spring Boot数据Geode序列化有关。让物体通量发挥作用的方法是使用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);
    }
  }
}