61610dea0e46a1df40cb6a221ad136f0200fd78e
[integration.git] /
1 /*-
2  * ============LICENSE_START=======================================================
3  * Simulator
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
10  * 
11  *      http://www.apache.org/licenses/LICENSE-2.0
12  * 
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=========================================================
19  */
20
21 package org.onap.netconfsimulator.websocket.message;
22
23 import java.io.IOException;
24 import javax.websocket.EncodeException;
25 import javax.websocket.RemoteEndpoint;
26 import org.apache.kafka.clients.consumer.ConsumerRecord;
27 import org.onap.netconfsimulator.kafka.model.KafkaMessage;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
30 import org.springframework.kafka.listener.MessageListener;
31
32 public class NetconfMessageListener implements MessageListener<String, String> {
33
34     private static final Logger LOGGER = LoggerFactory.getLogger(NetconfMessageListener.class);
35     private RemoteEndpoint.Basic remoteEndpoint;
36
37     public NetconfMessageListener(RemoteEndpoint.Basic remoteEndpoint) {
38         this.remoteEndpoint = remoteEndpoint;
39     }
40
41     @Override
42     public void onMessage(ConsumerRecord<String, String> message) {
43         LOGGER.debug("Attempting to send message to {}", remoteEndpoint);
44         try {
45             remoteEndpoint
46                 .sendObject(new KafkaMessage(message.timestamp(), message.value()));
47         } catch (IOException | EncodeException exception) {
48             LOGGER.error("Error during sending message to remote", exception);
49         }
50     }
51 }