@Test public void testSubscribers() { MediatorConfigurationSupport support = create("subscriberSubscriberOfMessage"); MediatorConfigurationSupport.ValidationOutput output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSubscriberOfPayload"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSubscriberBuilderOfMessage"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("subscriberSubscriberBuilderOfPayload"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isTrue(); assertThatThrownBy(() -> create("subscriberSinkOfMessage").validate(Shape.SUBSCRIBER, null)) .isInstanceOf(DefinitionException.class); support = create("subscriberSinkOfPayload"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSinkOfMessageCompletionStage"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSinkOfPayloadCompletionStage"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSinkOfMessageUni"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSinkOfPayloadUni"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSinkOfRawMessageCompletionStage"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(null); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("subscriberSinkOfWildcardMessageCompletionStage"); output = support.validate(Shape.SUBSCRIBER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.NONE); assertThat(output.getIngestedPayloadType()).isEqualTo(null); assertThat(output.getUseBuilderTypes()).isFalse(); }
public ValidationOutput validate(Shape shape, Acknowledgment.Strategy acknowledgment) { switch (shape) { case SUBSCRIBER: return validateSubscriber(); case PUBLISHER: return validatePublisher(); case PROCESSOR: return validateProcessor(acknowledgment); case STREAM_TRANSFORMER: return validateStreamTransformer(acknowledgment); default: throw ex.illegalStateExceptionForValidate(shape); } }
MediatorConfigurationSupport { public ValidationOutput validate(Shape shape, Acknowledgment.Strategy acknowledgment) { switch (shape) { case SUBSCRIBER: return validateSubscriber(); case PUBLISHER: return validatePublisher(); case PROCESSOR: return validateProcessor(acknowledgment); case STREAM_TRANSFORMER: return validateStreamTransformer(acknowledgment); default: throw ex.illegalStateExceptionForValidate(shape); } } }
MediatorConfigurationSupport { public ValidationOutput validate(Shape shape, Acknowledgment.Strategy acknowledgment) { switch (shape) { case SUBSCRIBER: return validateSubscriber(); case PUBLISHER: return validatePublisher(); case PROCESSOR: return validateProcessor(acknowledgment); case STREAM_TRANSFORMER: return validateStreamTransformer(acknowledgment); default: throw ex.illegalStateExceptionForValidate(shape); } } MediatorConfigurationSupport(String methodAsString, Class<?> returnType, Class<?>[] parameterTypes,
            GenericTypeAssignable returnTypeAssignable, GenericTypeAssignable firstMethodParamTypeAssignable); }
MediatorConfigurationSupport { public ValidationOutput validate(Shape shape, Acknowledgment.Strategy acknowledgment) { switch (shape) { case SUBSCRIBER: return validateSubscriber(); case PUBLISHER: return validatePublisher(); case PROCESSOR: return validateProcessor(acknowledgment); case STREAM_TRANSFORMER: return validateStreamTransformer(acknowledgment); default: throw ex.illegalStateExceptionForValidate(shape); } } MediatorConfigurationSupport(String methodAsString, Class<?> returnType, Class<?>[] parameterTypes,
            GenericTypeAssignable returnTypeAssignable, GenericTypeAssignable firstMethodParamTypeAssignable); Shape determineShape(List<?> incomingValue, Object outgoingValue); Acknowledgment.Strategy processSuppliedAcknowledgement(List<?> incomings,
            Supplier<Acknowledgment.Strategy> supplier); ValidationOutput validate(Shape shape, Acknowledgment.Strategy acknowledgment); Acknowledgment.Strategy processDefaultAcknowledgement(Shape shape,
            MediatorConfiguration.Consumption consumption, MediatorConfiguration.Production production); Merge.Mode processMerge(List<?> incomings, Supplier<Merge.Mode> supplier); Integer processBroadcast(Object outgoing, Supplier<Integer> supplier); void validateBlocking(ValidationOutput validationOutput); }
MediatorConfigurationSupport { public ValidationOutput validate(Shape shape, Acknowledgment.Strategy acknowledgment) { switch (shape) { case SUBSCRIBER: return validateSubscriber(); case PUBLISHER: return validatePublisher(); case PROCESSOR: return validateProcessor(acknowledgment); case STREAM_TRANSFORMER: return validateStreamTransformer(acknowledgment); default: throw ex.illegalStateExceptionForValidate(shape); } } MediatorConfigurationSupport(String methodAsString, Class<?> returnType, Class<?>[] parameterTypes,
            GenericTypeAssignable returnTypeAssignable, GenericTypeAssignable firstMethodParamTypeAssignable); Shape determineShape(List<?> incomingValue, Object outgoingValue); Acknowledgment.Strategy processSuppliedAcknowledgement(List<?> incomings,
            Supplier<Acknowledgment.Strategy> supplier); ValidationOutput validate(Shape shape, Acknowledgment.Strategy acknowledgment); Acknowledgment.Strategy processDefaultAcknowledgement(Shape shape,
            MediatorConfiguration.Consumption consumption, MediatorConfiguration.Production production); Merge.Mode processMerge(List<?> incomings, Supplier<Merge.Mode> supplier); Integer processBroadcast(Object outgoing, Supplier<Integer> supplier); void validateBlocking(ValidationOutput validationOutput);  }