core/core-infrastructure/src/main/java/org/onap/policy/apex/core/infrastructure/messaging/impl/ws/client/MessagingClient.java

   1 /*-
   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
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.apex.core.infrastructure.messaging.impl.ws.client;
  22
  23 import java.net.URI;
  24 import org.java_websocket.WebSocket;
  25 import org.onap.policy.apex.core.infrastructure.messaging.MessageHolder;
  26 import org.onap.policy.apex.core.infrastructure.messaging.MessagingService;
  27 import org.onap.policy.apex.core.infrastructure.messaging.util.MessagingUtils;
  28 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  29
  30 /**
  31  * The Class MessagingClient is the class that wraps web socket handling, message sending, and
  32  * message reception on the client side of a web socket in Apex.
  33  *
  34  * @author Sajeevan Achuthan (sajeevan.achuthan@ericsson.com)
  35  * @param <M> the generic type
  36  */
  37 public class MessagingClient<M> extends InternalMessageBusClient<M> implements MessagingService<M> {
  38     // The length of time to wait for a connection to a web socket server before aborting
  39     private static final int CONNECTION_TIMEOUT_TIME_MS = 3000;
  40
  41     // The length of time to wait before checking if a connection to a web socket server has worked
  42     // or not
  43     private static final int CONNECTION_TRY_INTERVAL_MS = 100;
  44
  45     /**
  46      * Constructor of this class, uses its {@link InternalMessageBusClient} superclass to set up the
  47      * web socket and handle incoming message forwarding.
  48      *
  49      * @param serverUri The URI of the service
  50      */
  51     public MessagingClient(final URI serverUri) {
  52         // Call the super class to create the web socket and set up received message forwarding
  53         super(serverUri);
  54     }
  55
  56     /**
  57      * {@inheritDoc}.
  58      */
  59     @Override
  60     public void stopConnection() {
  61         // Stop message reception in the super class
  62         super.stopListener();
  63
  64         // Close the web socket
  65         final WebSocket connection = super.getConnection();
  66         if (connection != null && connection.isOpen()) {
  67             connection.closeConnection(0, "");
  68         }
  69         this.close();
  70     }
  71
  72     /**
  73      * {@inheritDoc}.
  74      */
  75     @Override
  76     public void startConnection() {
  77         // Open the web socket
  78         final WebSocket connection = super.getConnection();
  79
  80         if (connection == null) {
  81             throw new IllegalStateException("Could not connect to the server");
  82         }
  83         if (!connection.isOpen()) {
  84             connect();
  85         }
  86
  87         if (!waitforConnection(connection)) {
  88             throw new IllegalStateException("Could not connect to the server");
  89         }
  90     }
  91
  92     /**
  93      * This method waits for the timeout value for the client to connect to the web socket server.
  94      *
  95      * @param connection the connection to wait on
  96      * @return true, if successful
  97      */
  98     private boolean waitforConnection(final WebSocket connection) {
  99         // The total time we have before timeout
 100         int timeoutMsCounter = CONNECTION_TIMEOUT_TIME_MS;
 101
 102         // Check the connection state
 103         do {
 104             switch (connection.getReadyState()) {
 105                 case NOT_YET_CONNECTED:
 106                 case CONNECTING:
 107                 case CLOSING:
 108                     // Not connected yet so wait for the try interval
 109                     ThreadUtilities.sleep(CONNECTION_TRY_INTERVAL_MS);
 110                     timeoutMsCounter -= CONNECTION_TRY_INTERVAL_MS;
 111                     break;
 112                 case OPEN:
 113                     // Connection is open, happy days
 114                     return true;
 115                 case CLOSED:
 116                     // Connection is closed, bah
 117                     return false;
 118                 default:
 119                     break;
 120             }
 121         } while (timeoutMsCounter > 0);
 122         // While the timeout value has not expired
 123
 124         // We have timed out
 125         return false;
 126     }
 127
 128     /**
 129      * {@inheritDoc}.
 130      */
 131     @Override
 132     public void send(final MessageHolder<M> commands) {
 133         // Get the connection and send the message
 134         final WebSocket connection = super.getConnection();
 135         connection.send(MessagingUtils.serializeObject(commands));
 136     }
 137
 138     /**
 139      * {@inheritDoc}.
 140      */
 141     @Override
 142     public void send(final String messageString) {
 143         final WebSocket connection = super.getConnection();
 144         connection.send(messageString);
 145     }
 146
 147     /**
 148      * {@inheritDoc}.
 149      */
 150     @Override
 151     public boolean isStarted() {
 152         return getConnection().isOpen();
 153     }
 154 }