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.simulators.xnf.impl
22 import com.google.protobuf.ByteString
23 import org.onap.dcae.collectors.veshv.domain.PayloadWireFrameMessage
24 import org.onap.dcae.collectors.veshv.simulators.xnf.api.MessageGenerator
25 import org.onap.dcae.collectors.veshv.simulators.xnf.config.MessageParameters
26 import org.onap.ves.VesEventV5.VesEvent
27 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
28 import reactor.core.publisher.Flux
29 import reactor.core.publisher.Mono
30 import javax.json.JsonObject
33 * @author Jakub Dudycz <jakub.dudycz@nokia.com>
36 internal class MessageGeneratorImpl(private val payloadGenerator: PayloadGenerator) : MessageGenerator {
38 override fun createMessageFlux(messageParameters: MessageParameters): Flux<PayloadWireFrameMessage> =
39 Mono.fromCallable { createMessage(messageParameters.commonEventHeader) }.let {
40 if (messageParameters.amount < 0)
43 it.repeat(messageParameters.amount)
46 fun parseCommonHeader(json: JsonObject): CommonEventHeader = CommonEventHeader.newBuilder()
47 .setVersion(json.getString("version"))
48 .setDomain(CommonEventHeader.Domain.forNumber(json.getInt("domain")))
49 .setSequence(json.getInt("sequence"))
50 .setPriority(CommonEventHeader.Priority.forNumber(json.getInt("priority")))
51 .setEventId(json.getString("eventId"))
52 .setEventName(json.getString("eventName"))
53 .setEventType(json.getString("eventType"))
54 .setStartEpochMicrosec(json.getJsonNumber("startEpochMicrosec").longValue())
55 .setLastEpochMicrosec(json.getJsonNumber("lastEpochMicrosec").longValue())
56 .setNfNamingCode(json.getString("nfNamingCode"))
57 .setNfcNamingCode(json.getString("nfcNamingCode"))
58 .setReportingEntityId(json.getString("reportingEntityId"))
59 .setReportingEntityName(ByteString.copyFromUtf8(json.getString("reportingEntityName")))
60 .setSourceId(ByteString.copyFromUtf8(json.getString("sourceId")))
61 .setSourceName(json.getString("sourceName"))
65 private fun createMessage(commonHeader: CommonEventHeader): PayloadWireFrameMessage =
66 PayloadWireFrameMessage(vesMessageBytes(commonHeader))
69 private fun vesMessageBytes(commonHeader: CommonEventHeader): ByteArray =
71 .setCommonEventHeader(commonHeader)
72 .setHvRanMeasFields(payloadGenerator.generatePayload().toByteString())
77 val INSTANCE = MessageGeneratorImpl(PayloadGenerator())