Пользовательский компонентный компонент Apache Camel не вызывает другие процессоры

Question

Я написал пользовательский компонент в Apache Camel .Верблюд успешно создал своего потребителя и использовал мой URI, но не вызвал ни одного процессора.Вот мой фрагмент кода потребителя (в Kotlin):

class SoroushBotConsumer (private val endpoint: MyEndpoint, processor: Processor) : DefaultConsumer(endpoint, processor) {
    val objectMapper:ObjectMapper = ObjectMapper();
    init {
        startListening()
    }
    private fun startListening() {
        val client = ClientBuilder.newBuilder().register(SseFeature::class.java).build()
        val target = client.target("MY_URL"))
        while(true){
            var e: EventInput?  target.request().get(EventInput::class.java)!!

            val inboundEvent = e.read()
            val exchange = endpoint.createExchange()
            exchange.getIn().body = objectMapper.readValue(inboundEvent.rawData,MessageModel::class.java)
            try {
                processor.process(exchange)
            } catch (e: Exception) {
                if (exchange.exception != null) {
                exceptionHandler.handleException("Error processing exchange",exchange, exchange.exception)
            }
        }
    }
}

все отлично работает в Consumer, но процессор не выполняется.вот как я создаю маршрут!

var context = DefaultCamelContext()
context.addRoutes(object : RouteBuilder() {
    override fun configure() {
        from("myapp://getMessage/).process{
            println(it.getIn())
        }.to("myapp://sendMessage/")
    }
})
context.start();
Thread.sleep(100000);
context.stop();

он не вызывает процесс и не создает моего производителя.(он даже не вызывает MyEndpoint::createProducer())

, когда я заменяю свой оператор from другой конечной точкой, такой как file, все работает нормально.

Обновление: когда я расширяю моего потребителя с ScheduledPollConsumer Реализуем pull метод, все идет хорошо.

Answer 1

Это не очень хорошая идея иметь бесконечный цикл в методе doStart, где вы будете захватывать текущий поток, который затем никогда не завершится.Вместо этого вы должны настроить фоновый поток, который выполняет это задание, и из doStart вы можете настроить этот поток и позволить ему работать.Другими словами, то, как компонент «получает» сообщения, на 100% зависит от компонента, поскольку каждое из них имеет свой собственный путь.А в методе doStop у вас есть логика для остановки фонового потока и очистки любых ваших ресурсов.

Answer 2

Его, потому что мы должны закончить конструктор потребителя и написать логику получения сообщений в doStart()

class SoroushBotConsumer (private val endpoint: MyEndpoint, processor: Processor) : DefaultConsumer(endpoint, processor) {
    val objectMapper:ObjectMapper = ObjectMapper();
    override fun doStart() {
        val client = ClientBuilder.newBuilder().register(SseFeature::class.java).build()
        val target = client.target("MY_URL"))
        while(true){
            var e: EventInput?  target.request().get(EventInput::class.java)!!

            val inboundEvent = e.read()
            val exchange = endpoint.createExchange()
            exchange.getIn().body = objectMapper.readValue(inboundEvent.rawData,MessageModel::class.java)
            try {
                processor.process(exchange)
            } catch (e: Exception) {
                if (exchange.exception != null) {
                    exceptionHandler.handleException("Error processing exchange",exchange, exchange.exception)
                }
            }
        }
    }
}