2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2016-2018 Ericsson. All rights reserved.
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.apex.core.infrastructure.messaging.impl.ws.server;
23 import java.net.InetSocketAddress;
24 import java.nio.ByteBuffer;
26 import org.java_websocket.WebSocket;
27 import org.onap.policy.apex.core.infrastructure.messaging.MessageListener;
28 import org.onap.policy.apex.core.infrastructure.messaging.MessagingService;
29 import org.onap.policy.apex.core.infrastructure.messaging.impl.ws.RawMessageHandler;
30 import org.onap.policy.apex.core.infrastructure.messaging.impl.ws.messageblock.MessageBlockHandler;
31 import org.onap.policy.apex.core.infrastructure.messaging.impl.ws.messageblock.RawMessageBlock;
32 import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
33 import org.slf4j.ext.XLogger;
34 import org.slf4j.ext.XLoggerFactory;
37 * The Class InternalMessageBusServer handles the server side of a web socket and handles the callback mechanism used to
38 * receive messages on the web socket.
40 * @author Sajeevan Achuthan (sajeevan.achuthan@ericsson.com)
41 * @param <M> the generic type
43 abstract class InternalMessageBusServer<M> extends WebSocketServerImpl implements MessagingService<M> {
44 // Logger for this class
45 private static final XLogger LOGGER = XLoggerFactory.getXLogger(InternalMessageBusServer.class);
47 private static final int THREAD_FACTORY_STACK_SIZE = 256;
49 // Name of the event bus.
50 private static final String RAW_EVENT_BUS = "Raw-Event-Bus";
52 // This instance handles the raw data received from the web socket
53 private final RawMessageHandler<M> rawMessageHandler = new RawMessageHandler<>();
55 // The message block handler to which to pass messages coming in on this client
56 private MessageBlockHandler<M> messageBlockHandler = null;
58 // The raw message handler uses a thread to process incoming events off a queue, this class owns and controls that
59 // thread. These fields hold the thread and
60 // the thread factory for creating threads.
61 private ApplicationThreadFactory threadFactory =
62 new ApplicationThreadFactory("ws-server-thread", THREAD_FACTORY_STACK_SIZE);
63 private Thread forwarderThread = null;
66 * Construct the class and start the forwarding thread for received messages.
68 * @param address the address of the server machine
70 protected InternalMessageBusServer(final InetSocketAddress address) {
71 // Call the super class to create the web socket
73 LOGGER.entry(address.getAddress().getHostAddress() + ":" + address.getPort());
75 // Create the data handler for forwarding messages
76 messageBlockHandler = new MessageBlockHandler<>(RAW_EVENT_BUS);
77 messageBlockHandler.registerMessageHandler(rawMessageHandler);
79 // Create the thread that manages the queue in the data handler
80 forwarderThread = threadFactory.newThread(rawMessageHandler);
81 forwarderThread.start();
87 * Callback for binary messages received from the remote host.
89 * @param webSocket the web socket on which the raw message was received
90 * @param rawMessage the received raw message
91 * @see #onMessage(WebSocket, String)
94 public void onMessage(final WebSocket webSocket, final ByteBuffer rawMessage) {
95 messageBlockHandler.post(new RawMessageBlock(rawMessage, webSocket));
102 public void onMessage(final WebSocket webSocket, final String stringMessage) {
103 messageBlockHandler.post(stringMessage);
107 * Register a subscriber class to the raw message handler.
109 * @param subscriber the subscriber
112 public void addMessageListener(final MessageListener<M> subscriber) {
113 rawMessageHandler.registerDataForwarder(subscriber);
117 * Removes the message listener.
119 * @param subscriber the subscriber
122 public void removeMessageListener(final MessageListener<M> subscriber) {
123 rawMessageHandler.unRegisterDataForwarder(subscriber);
127 * Stop the thread handling message forwarding.
129 protected void stopListener() {
130 rawMessageHandler.shutdown();