this.cause = cause;
}
+ public StreamParsingException(String message) {
+ this(new StreamParserError(message));
+ }
+
public StreamParserError cause() {
return cause;
}
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.RawDataStream;
/**
+ * Extract streams from the application configuration represented as GSON JsonObject.
+ *
+ * Example input:
+ * <pre>
+ * {
+ * "application_config_1": "value_1,
+ * ...
+ * "streams_publishes": {
+ * "stream1": {
+ * "type": "message_router",
+ * "dmaap_info": {
+ * ...
+ * }
+ * },
+ * "stream2": {
+ * "type": "data_router",
+ * "dmaap_info": {
+ * ...
+ * }
+ * }
+ * },
+ * "streams_subscribes": {
+ * "stream3": {
+ * "type": "message_router",
+ * "dmaap_info": {
+ * ...
+ * }
+ * },
+ * }
+ * }
+ * </pre>
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
* @since 1.1.4
*/
private DataStreams() {
}
+ /**
+ * <p>
+ * Extracts sources from application configuration. Parses <em>streams_subscribes</em> subtree.
+ * </p>
+ *
+ * <p>
+ * For sample input it will yield:
+ * </p>
+ *
+ * <pre>
+ * [
+ * RawDataStream{
+ * name="stream3"
+ * type="message_router"
+ * direction=SOURCE
+ * descriptor=JsonObject{
+ * type: "message_router",
+ * dmaap_info: {
+ * ...
+ * }
+ * }
+ * }
+ * ]
+ * </pre>
+ *
+ * @param rootJson - the full application configuration
+ * @return io.vavr.collection.Stream of data streams
+ */
public static Stream<RawDataStream<JsonObject>> namedSources(JsonObject rootJson) {
return createCollectionOfStreams(rootJson, DataStreamDirection.SOURCE);
}
+
+ /**
+ * <p>
+ * Extracts sinks from application configuration. Parses <em>streams_publishes</em> subtree.
+ * </p>
+ *
+ * <p>
+ * For sample input it will yield:
+ * </p>
+ *
+ * <pre>
+ * [
+ * RawDataStream{
+ * name="stream1"
+ * type="message_router"
+ * direction=SINK
+ * descriptor=JsonObject{
+ * type: "message_router",
+ * dmaap_info: {
+ * ...
+ * }
+ * }
+ * },
+ * RawDataStream{
+ * name="stream2"
+ * type="data_router"
+ * direction=SINK
+ * descriptor=JsonObject{
+ * type: "data_router"
+ * dmaap_info: {
+ * ...
+ * }
+ * }
+ * }
+ * ]
+ * </pre>
+ *
+ * @param rootJson - the full application configuration
+ * @return io.vavr.collection.Stream of data streams
+ */
public static Stream<RawDataStream<JsonObject>> namedSinks(JsonObject rootJson) {
return createCollectionOfStreams(rootJson, DataStreamDirection.SINK);
}
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.DataStream;
/**
+ * Represents parser taking GSON JsonObject as an input
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
* @since 1.1.4
*/
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap.*;
/**
+ * Factory methods for GSON-based {@code StreamParser}s
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
* @since 1.1.4
*/
private StreamFromGsonParsers() {
}
+ /**
+ * Creates a stream parser capable of parsing Kafka sinks.
+ *
+ * @return a stream parser
+ */
public static StreamFromGsonParser<KafkaSink> kafkaSinkParser() {
return KafkaSinkParser.create();
}
+ /**
+ * Creates a stream parser capable of parsing Kafka sources.
+ *
+ * @return a stream parser
+ */
public static StreamFromGsonParser<KafkaSource> kafkaSourceParser() {
return KafkaSourceParser.create();
}
+ /**
+ * Creates a stream parser capable of parsing DMaaP Message Router sinks.
+ *
+ * @return a stream parser
+ */
public static StreamFromGsonParser<MessageRouterSink> messageRouterSinkParser() {
return MessageRouterSinkParser.create();
}
+ /**
+ * Creates a stream parser capable of parsing DMaaP Message Router sources.
+ *
+ * @return a stream parser
+ */
public static StreamFromGsonParser<MessageRouterSource> messageRouterSourceParser() {
return MessageRouterSourceParser.create();
}
+ /**
+ * Creates a stream parser capable of parsing DMaaP Data Router sinks.
+ *
+ * @return a stream parser
+ */
public static StreamFromGsonParser<DataRouterSink> dataRouterSinkParser() {
return DataRouterSinkParser.create();
}
+ /**
+ * Creates a stream parser capable of parsing DMaaP Data Router sources.
+ *
+ * @return a stream parser
+ */
public static StreamFromGsonParser<DataRouterSource> dataRouterSourceParser() {
return DataRouterSourceParser.create();
}
import com.google.gson.JsonObject;
import io.vavr.collection.Stream;
import java.io.IOException;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParsingException;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.DataStreamDirection;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.ImmutableRawDataStream;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.RawDataStream;
String expectedType,
DataStreamDirection expectedDirection) {
if (!json.type().equals(expectedType)) {
- throw new IllegalArgumentException(
+ throw new StreamParsingException(
"Invalid stream type. Expected '" + expectedType + "', but was '" + json.type() + "'");
}
if (json.direction() != expectedDirection) {
- throw new IllegalArgumentException(
+ throw new StreamParsingException(
"Invalid stream direction. Expected '" + expectedDirection + "', but was '" + json.direction()
+ "'");
}
import java.io.Reader;
import java.util.Map.Entry;
import java.util.stream.Collectors;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParsingException;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.dmaap.mr.GsonAdaptersMessageRouterDmaapInfo;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.streams.gson.kafka.GsonAdaptersKafkaInfo;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.GsonAdaptersAafCredentials;
public static JsonElement requiredChild(JsonObject parent, String childName) {
return optionalChild(parent, childName)
- .getOrElseThrow(() -> new IllegalArgumentException(
+ .getOrElseThrow(() -> new StreamParsingException(
"Could not find sub-node '" + childName + "'. Actual sub-nodes: "
+ stringifyChildrenNames(parent)));
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
/**
+ * Represents the AAF Credentials. Currently it contains only user name and password.
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
@Value.Immutable
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
/**
+ * Represents a named data stream.
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
public interface DataStream {
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams;
/**
+ * The direction of the stream, ie. whether it's input ({@code SOURCE}) or output ({@code SINK}) stream.
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since March 2019
+ * @since 1.1.4
*/
public enum DataStreamDirection {
this.configurationKey = configurationKey;
}
+ /**
+ * The configuration key under which the single stream definitions should reside.
+ *
+ * @return the configuration key
+ */
public String configurationKey() {
return configurationKey;
}
import org.immutables.value.Value;
/**
+ * Represents a raw/uninterpreted data stream.
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since March 2019
+ * @since 1.1.4
+ * @param <T> type of raw data, eg. JsonObject
*/
@Value.Immutable
public interface RawDataStream<T> {
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
/**
- * AKA PublishStream
+ * Represents an output stream, ie. one of objects in <em>streams_publishes</em> array from application configuration.
+ * Application can put data to this stream.
*
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
public interface SinkStream extends DataStream {
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
/**
- * AKA SubscribeStream
+ * Represents an input stream, ie. one of objects in <em>streams_subscribes</em> array from application configuration.
+ * Application can read data from this stream.
*
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
public interface SourceStream extends DataStream {
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
public interface DataRouter {
+ /**
+ * DCAE location for the publisher, used to set up routing.
+ */
@SerializedName("location")
@Nullable String location();
+ /**
+ * Username
+ * <ul>
+ * <li>Data Router uses to authenticate to the subscriber when delivering files OR</li>
+ * <li>the publisher uses to authenticate to Data Router.</li>
+ * </ul>
+ */
@SerializedName("username")
@Nullable String username();
+ /**
+ * Password
+ * <ul>
+ * <li>Data Router uses to authenticate to the subscriber when delivering files OR</li>
+ * <li>the publisher uses to authenticate to Data Router.</li>
+ * </ul>
+ */
@SerializedName("password")
@Nullable String password();
}
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@Gson.TypeAdapters
@ExperimentalApi
@Value.Immutable
public interface DataRouterSink extends DataRouter, SinkStream {
+ /**
+ * URL to which the publisher makes Data Router publish requests.
+ */
@SerializedName("publish_url")
String publishUrl();
+ /**
+ * Publisher id in Data Router
+ */
@SerializedName("publisher_id")
@Nullable String publisherId();
+ /**
+ * URL from which log data for the feed can be obtained.
+ */
@SerializedName("log_url")
@Nullable String logUrl();
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@Gson.TypeAdapters
@ExperimentalApi
@Value.Immutable
public interface DataRouterSource extends DataRouter, SourceStream {
- // in future, since crucial need to be verified if could be nullable
+ /**
+ * URL to which the Data Router should deliver files.
+ */
+ // TODO: since crucial, we need to verify if it should be non-null
@SerializedName("delivery_url")
@Nullable String deliveryUrl();
+ /**
+ * Subscriber id in Data Router.
+ */
@SerializedName("subscriber_id")
@Nullable String subscriberId();
}
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
import static io.vavr.Predicates.not;
@ExperimentalApi
public interface Kafka {
+ /**
+ * Kafka bootstrap servers as defined in Kafka client documentation under <em>bootstrap.servers</em> configuration
+ * key.
+ */
String bootstrapServers();
+ /**
+ * The name of the topic where application should publish or subscribe for the messages.
+ */
String topicName();
+ /**
+ * The credentials to use when authenticating to Kafka cluster or null when connection should be unauthenticated.
+ */
@Nullable AafCredentials aafCredentials();
+ /**
+ * AAF client role that’s requesting publish or subscribe access to the topic.
+ */
@Nullable String clientRole();
+ /**
+ * Client id for given AAF client.
+ */
@Nullable String clientId();
+ /**
+ * The limit on the size of message published to/subscribed from the topic. Can be used to set Kafka client
+ * <em>max.request.size</em> configuration property.
+ */
@Value.Default
default int maxPayloadSizeBytes() {
return 1024 * 1024;
}
+ /**
+ * The {@code bootstrapServers} converted to the list of servers' addresses.
+ */
@Value.Derived
default List<String> bootstrapServerList() {
return List.of(bootstrapServers().split(",")).filter(not(String::isEmpty));
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
import org.immutables.value.Value;
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
import org.immutables.value.Value;
@Value.Immutable
public interface KafkaSource extends Kafka, SourceStream {
+ /**
+ * A unique string that identifies the consumer group this consumer belongs to as defined in Kafka consumer
+ * configuration key <em>group.id</em>.
+ */
@Nullable String consumerGroupId();
}
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
-
import com.google.gson.annotations.SerializedName;
import org.jetbrains.annotations.Nullable;
import org.onap.dcaegen2.services.sdk.rest.services.annotations.ExperimentalApi;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
public interface MessageRouter {
+ /**
+ * URL for accessing the topic to publish or receive events.
+ */
@SerializedName("topic_url")
String topicUrl();
+ /**
+ * AAF client role that’s requesting publish or subscribe access to the topic.
+ */
@SerializedName("client_role")
@Nullable String clientRole();
+ /**
+ * Client id for given AAF client.
+ */
@SerializedName("client_id")
@Nullable String clientId();
+ /**
+ * DCAE location for the publisher or subscriber, used to set up routing.
+ */
@SerializedName("location")
@Nullable String location();
+ /**
+ * The AAF credentials.
+ */
@SerializedName("aaf_credentials")
@Nullable AafCredentials aafCredentials();
}
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
import org.immutables.value.Value;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
@Value.Immutable
* limitations under the License.
* ============LICENSE_END=====================================
*/
-
package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.streams.dmaap;
import org.immutables.value.Value;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @version 1.2.1
+ * @since 1.1.4
*/
@ExperimentalApi
@Value.Immutable
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.exceptions.StreamParsingException;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.DataStreams;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParser;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.streams.StreamFromGsonParsers;
// then
StepVerifier.create(result)
.expectErrorSatisfies(ex -> {
- assertThat(ex).isInstanceOf(IllegalArgumentException.class);
+ assertThat(ex).isInstanceOf(StreamParsingException.class);
assertThat(ex).hasMessageContaining("Invalid stream type");
assertThat(ex).hasMessageContaining("message_router");
assertThat(ex).hasMessageContaining("kafka");
Code Review / dcaegen2 / services / sdk.git / commitdiff