2 * ============LICENSE_START=======================================================
3 * DCAEGEN2-SERVICES-SDK
4 * ================================================================================
5 * Copyright (C) 2019 Nokia. All rights reserved.
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.dcaegen2.services.sdk.services.hvves.client.producer.impl;
22 import static org.assertj.core.api.Assertions.assertThat;
23 import static org.mockito.ArgumentMatchers.any;
24 import static org.mockito.Mockito.mock;
25 import static org.mockito.Mockito.times;
26 import static org.mockito.Mockito.verify;
27 import static org.mockito.Mockito.when;
28 import static org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl.utils.VesEvents.defaultVesEvent;
30 import io.netty.buffer.ByteBuf;
31 import io.netty.buffer.ByteBufAllocator;
32 import io.netty.buffer.Unpooled;
33 import io.vavr.control.Try;
34 import org.junit.jupiter.api.BeforeEach;
35 import org.junit.jupiter.api.Test;
36 import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl.encoders.EncodersFactory;
37 import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl.encoders.ProtobufEncoder;
38 import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl.encoders.WireFrameEncoder;
39 import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.options.WireFrameVersion;
40 import org.onap.ves.VesEventOuterClass.VesEvent;
41 import reactor.core.publisher.Flux;
43 import java.nio.ByteBuffer;
46 * @author <a href="mailto:jakub.dudycz@nokia.com">Jakub Dudycz</a>
48 public class ProducerCoreTest {
50 private ProducerCore producerCore;
51 private EncodersFactory encodersFactoryMock;
52 private WireFrameVersion wireFrameVersion;
56 encodersFactoryMock = mock(EncodersFactory.class);
57 wireFrameVersion = mock(WireFrameVersion.class);
58 producerCore = new ProducerCore(encodersFactoryMock, wireFrameVersion);
62 public void encode_should_encode_message_stream_to_wire_frame() {
63 final WireFrameEncoder wireFrameEncoder = mock(WireFrameEncoder.class);
64 final ProtobufEncoder protobufEncoder = mock(ProtobufEncoder.class);
65 final ByteBuffer protoBuffer = ByteBuffer.wrap(new byte[3]);
66 final Try<ByteBuf> wireFrameBuffer = Try.success(Unpooled.copiedBuffer(new byte[5]));
68 when(protobufEncoder.encode(any(VesEvent.class))).thenReturn(protoBuffer);
69 when(wireFrameEncoder.encode(protoBuffer)).thenReturn(wireFrameBuffer);
70 when(encodersFactoryMock.createProtobufEncoder()).thenReturn(protobufEncoder);
71 when(encodersFactoryMock.createWireFrameEncoder(ByteBufAllocator.DEFAULT, wireFrameVersion)).
72 thenReturn(wireFrameEncoder);
75 final int messageStreamSize = 2;
76 final Flux<VesEvent> messages = Flux.just(defaultVesEvent()).repeat(messageStreamSize - 1);
79 final ByteBuf lastMessage = producerCore.encode(messages, ByteBufAllocator.DEFAULT).blockLast();
82 verify(encodersFactoryMock).createProtobufEncoder();
83 verify(encodersFactoryMock).createWireFrameEncoder(ByteBufAllocator.DEFAULT, wireFrameVersion);
84 verify(protobufEncoder, times(messageStreamSize)).encode(any(VesEvent.class));
85 verify(wireFrameEncoder, times(messageStreamSize)).encode(protoBuffer);
87 assertThat(lastMessage).isNotNull();
88 assertThat(lastMessage).isEqualTo(wireFrameBuffer.get());