class ImmutableSource implements Source {

    public static Source fromJson(final JsonObject jsonObject, final int index) {
        final Set<String> readSources = jsonObject.getValue(JsonFields.ADDRESSES)
                .map(array -> array.stream()
                        .map(JsonValue::asString)
                        .collect(Collectors.toSet())).orElse(Collections.emptySet());
        final int readConsumerCount =
                jsonObject.getValue(JsonFields.CONSUMER_COUNT).orElse(DEFAULT_CONSUMER_COUNT);
        final Integer readQos = jsonObject.getValue(JsonFields.QOS).orElse(null);
        final JsonArray authContext = jsonObject.getValue(JsonFields.AUTHORIZATION_CONTEXT)
                .orElseGet(() -> JsonArray.newBuilder().build());
        final List<AuthorizationSubject> authorizationSubjects = authContext.stream()
                .filter(JsonValue::isString)
                .map(JsonValue::asString)
                .map(AuthorizationSubject::newInstance)
                .collect(Collectors.toList());
        final AuthorizationContext readAuthorizationContext =
                AuthorizationModelFactory.newAuthContext(DittoAuthorizationContextType.PRE_AUTHENTICATED_CONNECTION,
                        authorizationSubjects);

        final Enforcement readEnforcement =
                jsonObject.getValue(JsonFields.ENFORCEMENT).map(ImmutableEnforcement::fromJson).orElse(null);

        final FilteredAcknowledgementRequest readAcknowledgementRequests =
                jsonObject.getValue(JsonFields.ACKNOWLEDGEMENT_REQUESTS)
                        .map(FilteredAcknowledgementRequest::fromJson)
                        .orElse(null);

        final HeaderMapping readHeaderMapping =
                jsonObject.getValue(JsonFields.HEADER_MAPPING).map(ImmutableHeaderMapping::fromJson).orElse(null);

        final PayloadMapping readPayloadMapping =
                jsonObject.getValue(JsonFields.PAYLOAD_MAPPING)
                        .map(ImmutablePayloadMapping::fromJson)
                        .orElse(ConnectivityModelFactory.emptyPayloadMapping());

        final boolean replyTargetEnabled =
                jsonObject.getValue(JsonFields.REPLY_TARGET_ENABLED).orElse(DEFAULT_REPLY_TARGET_ENABLED);

        final ReplyTarget readReplyTarget =
                jsonObject.getValue(JsonFields.REPLY_TARGET)
                        .flatMap(ImmutableReplyTarget::fromJsonOptional)
                        .orElse(null);

        return new Builder()
                .addresses(readSources)
                .qos(readQos)
                .authorizationContext(readAuthorizationContext)
                .consumerCount(readConsumerCount)
                .index(index)
                .enforcement(readEnforcement)
                .acknowledgementRequests(readAcknowledgementRequests)
                .headerMapping(readHeaderMapping)
                .payloadMapping(readPayloadMapping)
                .replyTargetEnabled(replyTargetEnabled)
                .replyTarget(readReplyTarget)
                .build();
    }

    private  ImmutableSource(final Builder builder);

}