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.main.impl
22 import com.google.protobuf.ByteString
23 import io.netty.buffer.ByteBuf
24 import io.netty.buffer.Unpooled
25 import org.onap.dcae.collectors.veshv.domain.WireFrame
26 import org.onap.ves.VesEventV5
27 import reactor.core.publisher.Flux
28 import reactor.core.publisher.Mono
31 * @author Jakub Dudycz <jakub.dudycz@nokia.com>
34 class MessageFactory {
37 const val DEFAULT_START_EPOCH: Long = 120034455
38 const val DEFAULT_LAST_EPOCH: Long = 120034455
41 fun createMessageFlux(amount: Int = 1): Flux<WireFrame> =
42 Mono.just(createMessage()).repeat(amount.toLong())
45 private fun createMessage(): WireFrame {
46 val commonHeader = VesEventV5.VesEvent.CommonEventHeader.newBuilder()
48 .setEventName("Sample event name")
49 .setDomain(VesEventV5.VesEvent.CommonEventHeader.Domain.HVRANMEAS)
50 .setEventId("Sample event Id")
51 .setSourceName("Sample Source")
52 .setReportingEntityName(ByteString.copyFromUtf8("Sample byte String"))
53 .setPriority(VesEventV5.VesEvent.CommonEventHeader.Priority.MEDIUM)
54 .setStartEpochMicrosec(DEFAULT_START_EPOCH)
55 .setLastEpochMicrosec(DEFAULT_LAST_EPOCH)
59 val payload = vesMessageBytes(commonHeader)
65 payloadSize = payload.readableBytes())
70 private fun vesMessageBytes(commonHeader: VesEventV5.VesEvent.CommonEventHeader): ByteBuf {
71 val msg = VesEventV5.VesEvent.newBuilder()
72 .setCommonEventHeader(commonHeader)
73 .setHvRanMeasFields(ByteString.copyFromUtf8("high volume data"))
76 return Unpooled.wrappedBuffer(msg.toByteArray())