Apache骆驼反应流抛出该流没有主动订阅

Question

我只是尝试使用骆驼反应流与Spring Boot反应器一起使用以下代码

package com.manning.camel.reactive;

import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

/**
 * A simple Camel route that triggers from a timer and calls a bean and prints to system out.
 * <p/>
 * Use <tt>@Component</tt> to make Camel auto-detect this route when starting.
 */
@RestController
public class MySpringBootRouter extends RouteBuilder {

  @Autowired
  private ProducerTemplate template;

  @Autowired
  private CamelReactiveStreamsService crss;

  @GetMapping
  public Mono<String> sayHi() {
    template.asyncSendBody("direct:works", "Hi");
    //return  Mono.from(crss.fromStream("greet", String.class));
    return Mono.from(crss.fromStream("greet", String.class));
  }

  @Override
  public void configure() {
    from("direct:works")
            .log("Fired")
            .to("reactive-streams:greet");
  }

}

，然后运行代码

java.lang.illegalstateexception：该流没有活动订阅

Answer 1

很长一段时间后，解决了错误，可以注意到路由器类逻辑有点更改了

@Slf4j
@Service
@AllArgsConstructor
public class MyService {
   final CamelContext context;

   @PostConstruct
   public void consumerData() {
     var rCamel = CamelReactiveStreams.get(context);
     var numbers = rCamel.fromStream("numbers", Integer.class);
     Flux.from(numbers).subscribe(e -> log.info("{}", e));
   }
}

@Component
@NoArgsConstructor
public class MyRouter extends RouteBuilder {


// Injects the Subscriber
@Autowired MyService service;

@Override
public void configure() {

  //onException(ReactiveStreamsNoActiveSubscriptionsException.class)
  //      .continued(true);

  from("timer://reactiveApp?fixedRate=true&period=2s")
        .transform(method(Random.class, "nextInt(100)"))
        //.log("${body}");
        .to("direct:message");

  from("direct:message")
        //.log("${body}")
        .to("reactive-streams:numbers");
  }
}