2 * ============LICENSE_START=======================================================
3 * dcaegen2-collectors-veshv
4 * ================================================================================
5 * Copyright (C) 2018 NOKIA
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
20 package org.onap.dcae.collectors.veshv.model
22 import arrow.core.Option
23 import org.onap.ves.VesEventOuterClass.CommonEventHeader
25 data class Routing(val routes: List<Route>) {
27 fun routeFor(commonHeader: CommonEventHeader): Option<Route> =
28 Option.fromNullable(routes.find { it.applies(commonHeader) })
31 data class Route(val domain: String, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int) {
33 fun applies(commonHeader: CommonEventHeader) = commonHeader.domain == domain
35 operator fun invoke(message: VesMessage): RoutedMessage =
36 RoutedMessage(targetTopic, partitioning(message.header), message)
44 fun routing(init: RoutingBuilder.() -> Unit): RoutingBuilder {
45 val conf = RoutingBuilder()
50 class RoutingBuilder {
51 private val routes: MutableList<RouteBuilder> = mutableListOf()
53 fun defineRoute(init: RouteBuilder.() -> Unit): RouteBuilder {
54 val rule = RouteBuilder()
60 fun build() = Routing(routes.map { it.build() }.toList())
65 private lateinit var domain: String
66 private lateinit var targetTopic: String
67 private lateinit var partitioning: (CommonEventHeader) -> Int
69 fun fromDomain(domain: String) {
73 fun toTopic(targetTopic: String) {
74 this.targetTopic = targetTopic
77 fun withFixedPartitioning(num: Int = 0) {
78 partitioning = { num }
81 fun build() = Route(domain, targetTopic, partitioning)