Conditional handler in integration flow

Question

I'm developing web hook notification service that allows clients to subscribe/unsubscribe to messages flowing through middle-ware and get notified about the messages (according to provided criteria) by posting the message payload to provided callback URL. The message delivery looks like this :

flowBuilder
    .enrichHeaders(e->e.header(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE,true))
    .handle(Http.outboundChannelAdapter(message-> {
              String subscriptionId = message.getHeaders().get(SUBSCRIPTION_ID_HEADER_NAME, String.class);
                return subscriptionsStore.get(UUID.fromString(subscriptionId)).getCallbackUrl(); //potential NPE if subscription was removed 

            },restTemplateBuilder.build())

.get()

As you can see, the implementation of uriFunction fetches the callback URL from subscriptionsStore by subscription id(part of the message header).

My question is about the situation where the client has already unsubscribed with his subscription id and I'm after the conditional handler.

I know that I can filter messages with subscription id are still present in subscription store, but this is not the proper solution, as client might unsubscribe between filter and handle operations still causing NRE in uriFunction.

Another solution is to enreach the header with callback URL and filter then by header having non-empty value, but I don't want to compromise neither header nor payload of original message.

I can think about another approach: to calculate the URI of non-existing subscriptions as some static value and add interceptor to RestTempalte to simulate the HTTP OK replay for this specific URI value...

So my question is about the proper way to handle this case by using the standard EIP or another Spring integration feature I'm not aware about...

Thanks

UPDATE

I've added the DedicatedMessage class that holds the context :

public static class DedicatedMessage  extends GenericMessage<Object> implements MessageDecorator{
        @Getter
        @Transient
        private Subscription subscription;


        public DedicatedMessage(Subscription subscription,Object payload,Map<String,Object> headers) {
            super(payload,headers);
            this.subscription = subscription;
        }

        @Override
        public Message<?> decorateMessage(Message<?> message) {
            return new DedicatedMessage (subscription,message.getPayload(),message.getHeaders());
        }
    }

and changed the flow as :

flowBuilder
   .enrichHeaders(e->e.header(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE,true))
  .handle((payload, headers) -> {
                        String subscriptionId = (String) headers.get(SUBSCRIPTION_ID_HEADER_NAME);
                        Subscription subscription = subscriptionsCache.get(UUID.fromString(subscriptionId));
                        return Optional.ofNullable(subscription)
                                .map(s->  new DedicatedMessage(s, payload,headers))
                                .orElse(null);
                    })
  .handle(Http.outboundChannelAdapter(message->((DedicatedMessage)message).getSubscription().getCallbackUrl()
                            ,restTemplateBuilder.build())
.get()

Any issues with this apporach ?

Gary Russell Gary Russell · Accepted Answer · 2017-11-05T17:15:05

I am not sure what your NRE abbreviation means, but you could throw a NoSuchSubscriptionException from your subscriptionsStore.get() method and then ignore/report that exception in an ExpressionEvaluatingRequestHandlerAdvice applied to the outbound channel adapter in its endpoint.advice() chain.

Conditional handler in integration flow

1 Answers