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);  }
@Test public void testPublishers() { MediatorConfigurationSupport support = create("publisherPublisherOfMessage"); MediatorConfigurationSupport.ValidationOutput output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherMultiOfMessage"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherPublisherOfPayload"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherMultiOfPayload"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherPublisherBuilderOfMessage"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("publisherPublisherBuilderOfPayload"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("publisherGeneratePayload"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.INDIVIDUAL_PAYLOAD); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherGenerateMessage"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.INDIVIDUAL_MESSAGE); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherGenerateCompletionStagePayload"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.COMPLETION_STAGE_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherGenerateCompletionStageMessage"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.COMPLETION_STAGE_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherGenerateUniPayload"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.UNI_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("publisherGenerateUniMessage"); output = support.validate(Shape.PUBLISHER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.NONE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.UNI_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isNull(); assertThat(output.getUseBuilderTypes()).isFalse(); }
@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(); }
@Test public void testProcessors() { MediatorConfigurationSupport support = create("processorProcessorOfMessage"); MediatorConfigurationSupport.ValidationOutput output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessorOfPayload"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessorBuilderOfMessage"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("processorProcessorBuilderOfPayload"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("processorPublisherOfMessage"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorPublisherOfPayload"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorPublisherBuilderOfMessage"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("processorPublisherBuilderOfPayload"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(String.class); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("processorProcessMessage"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.INDIVIDUAL_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessPayload"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.INDIVIDUAL_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessMessageCompletionStage"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.COMPLETION_STAGE_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessPayloadCompletionStage"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.COMPLETION_STAGE_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessMessageUni"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.UNI_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessPayloadUni"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.UNI_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessMessageUniRaw"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.UNI_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(null); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("processorProcessMessageUniWildcard"); output = support.validate(Shape.PROCESSOR, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.UNI_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(null); assertThat(output.getUseBuilderTypes()).isFalse(); }
@Test public void testStreamTransformers() { MediatorConfigurationSupport support = create("transformerPublisherOfMessage"); MediatorConfigurationSupport.ValidationOutput output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("transformerPublisherOfPayload"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("transformerMultiOfMessage"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("transformerMultiOfPayload"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("transformerPublisherBuilderOfMessage"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("transformerPublisherBuilderOfPayload"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(Person.class); assertThat(output.getUseBuilderTypes()).isTrue(); support = create("transformerPublisherOfMessageRaw"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(null); assertThat(output.getUseBuilderTypes()).isFalse(); support = create("transformerPublisherOfMessageWildcard"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_MESSAGE); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_MESSAGE); assertThat(output.getIngestedPayloadType()).isEqualTo(null); assertThat(output.getUseBuilderTypes()).isFalse(); assertThatThrownBy(() -> create("transformerPublisherOfPayloadRaw").validate(Shape.STREAM_TRANSFORMER, null)) .isInstanceOf(DefinitionException.class); support = create("transformerPublisherOfPayloadWildcard"); output = support.validate(Shape.STREAM_TRANSFORMER, null); assertThat(output.getConsumption()).isEqualTo(MediatorConfiguration.Consumption.STREAM_OF_PAYLOAD); assertThat(output.getProduction()).isEqualTo(MediatorConfiguration.Production.STREAM_OF_PAYLOAD); assertThat(output.getIngestedPayloadType()).isEqualTo(null); assertThat(output.getUseBuilderTypes()).isFalse(); }