@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();
}