Раскрытие RxMule

Я очень рад объявить  RxMule , мой последний проект с открытым исходным кодом. RxMule  предоставляет  реактивные расширения  для  Mule ESB через набор специфических привязок для  RxJava .

В течение нескольких лет я размышлял над идеей создания DSL для настройки Mule. Действительно, существует множество сокровищ уже существующих  транспортов  и  коннекторов  для Mule, что очень важно для всех, кто создает подключенные приложения (а сегодня это, вероятно, все). К сожалению, разработчики могут быть отстранены от использования DSL на основе XML, используемого для настройки Mule, и, таким образом, могут упустить возможность использовать все эти доступные качества.

По мере того, как Rx набирает обороты, все больше и больше разработчиков привыкли к его концепциям и примитивам. Имея это в виду и зная, что Mule является ядром платформы обработки событий, я понял, что вместо создания DSL, который будет имитировать артефакты XML (специфичные для Mule), я бы предпочел создать привязки, позволяющие использовать необходимые для Mule движущиеся части через Rx.

Таким образом,  RxMule  добавляет ряд классов в RxJava, которые позволяют наблюдать: 

• Множество входящих конечных точек от традиционных  транспортов , включая глобальные конечные точки и конечные точки, определенные URI,
• необработанные источники сообщений, такие как новый  коннектор прослушивателя HTTP ,
• Источники  сообщений Anypoint Connectors .

Короче говоря,  RxMule  позволяет создавать  Observable<MuleEvent> экземпляры из разных источников. MuleEvent  является то , что создает и процессы Mule. Он оборачивает  MuleMessage,  который содержит фактические данные и метаданные, которые обрабатываются.

Вы можете прочитать больше о структуре 
MuleMessage 
здесь .

Следующее демонстрирует асинхронный мост HTTP-Redis, который принимает только один запрос на удаленный IP: 

rxMule
    .observeEndpointAsync(new URI("http://localhost:8080/publish"))
    .distinct(
        muleEvent -> {
 
            final String remoteAddressAndPort =
                muleEvent
                    .getMessage()
                    .getInboundProperty(
                        "MULE_REMOTE_CLIENT_ADDRESS");
 
            return substringBefore(remoteAddressAndPort, ":");
        })
    .subscribe(
        asAction((MessageConsumer)
            muleEvent -> {
 
                redisModule.publish(
                    "http-requests",
                    false,
                    muleEvent.getMessageAsString(),
                    muleEvent);
 
                LOGGER.info("Published: {}", muleEvent);
    }));