git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2023 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.io.openhabcloud.internal;
  14
  15 import java.io.IOException;
  16 import java.net.MalformedURLException;
  17 import java.net.URI;
  18 import java.net.URISyntaxException;
  19 import java.net.URL;
  20 import java.net.URLEncoder;
  21 import java.nio.charset.StandardCharsets;
  22 import java.util.Iterator;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Optional;
  26 import java.util.Set;
  27 import java.util.concurrent.ConcurrentHashMap;
  28 import java.util.concurrent.ScheduledExecutorService;
  29 import java.util.concurrent.ScheduledFuture;
  30 import java.util.concurrent.TimeUnit;
  31 import java.util.concurrent.atomic.AtomicReference;
  32
  33 import org.eclipse.jdt.annotation.Nullable;
  34 import org.eclipse.jetty.client.HttpClient;
  35 import org.eclipse.jetty.client.api.Request;
  36 import org.eclipse.jetty.client.util.BytesContentProvider;
  37 import org.eclipse.jetty.http.HttpField;
  38 import org.eclipse.jetty.http.HttpFields;
  39 import org.eclipse.jetty.http.HttpMethod;
  40 import org.eclipse.jetty.http.HttpStatus;
  41 import org.eclipse.jetty.util.BufferUtil;
  42 import org.eclipse.jetty.util.URIUtil;
  43 import org.json.JSONException;
  44 import org.json.JSONObject;
  45 import org.openhab.core.OpenHAB;
  46 import org.openhab.core.common.ThreadPoolManager;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 import io.socket.backo.Backoff;
  51 import io.socket.client.IO;
  52 import io.socket.client.IO.Options;
  53 import io.socket.client.Manager;
  54 import io.socket.client.Socket;
  55 import io.socket.emitter.Emitter;
  56 import io.socket.engineio.client.Transport;
  57 import io.socket.engineio.client.transports.WebSocket;
  58 import io.socket.parser.Packet;
  59 import io.socket.parser.Parser;
  60 import okhttp3.OkHttpClient.Builder;
  61 import okhttp3.logging.HttpLoggingInterceptor;
  62 import okhttp3.logging.HttpLoggingInterceptor.Level;
  63
  64 /**
  65  * This class provides communication between openHAB and the openHAB Cloud service.
  66  * It also implements async http proxy for serving requests from user to
  67  * openHAB through the openHAB Cloud. It uses Socket.IO connection to connect to
  68  * the openHAB Cloud service and Jetty Http client to send local http requests to
  69  * openHAB.
  70  *
  71  * @author Victor Belov - Initial contribution
  72  * @author Kai Kreuzer - migrated code to new Jetty client and ESH APIs
  73  */
  74 public class CloudClient {
  75
  76     private static final long RECONNECT_MIN = 2_000;
  77
  78     private static final long RECONNECT_MAX = 60_000;
  79
  80     private static final double RECONNECT_JITTER = 0.75;
  81
  82     private static final long READ_TIMEOUT = 60_0000;
  83
  84     /*
  85      * Logger for this class
  86      */
  87     private final Logger logger = LoggerFactory.getLogger(CloudClient.class);
  88
  89     /*
  90      * This variable holds base URL for the openHAB Cloud connections
  91      */
  92     private final String baseURL;
  93
  94     /*
  95      * This variable holds openHAB's UUID for authenticating and connecting to the openHAB Cloud
  96      */
  97     private final String uuid;
  98
  99     /*
 100      * This variable holds openHAB's secret for authenticating and connecting to the openHAB Cloud
 101      */
 102     private final String secret;
 103
 104     /*
 105      * This variable holds local openHAB's base URL for connecting to the local openHAB instance
 106      */
 107     private final String localBaseUrl;
 108
 109     /*
 110      * This variable holds instance of Jetty HTTP client to make requests to local openHAB
 111      */
 112     private final HttpClient jettyClient;
 113
 114     /*
 115      * This map holds HTTP requests to local openHAB which are currently running
 116      */
 117     private final Map<Integer, Request> runningRequests = new ConcurrentHashMap<>();
 118
 119     /*
 120      * This variable indicates if connection to the openHAB Cloud is currently in an established state
 121      */
 122     private boolean isConnected;
 123
 124     /*
 125      * This variable holds instance of Socket.IO client class which provides communication
 126      * with the openHAB Cloud
 127      */
 128     private Socket socket;
 129
 130     /*
 131      * The protocol of the openHAB-cloud URL.
 132      */
 133     private String protocol = "https";
 134
 135     /*
 136      * This variable holds instance of CloudClientListener which provides callbacks to communicate
 137      * certain events from the openHAB Cloud back to openHAB
 138      */
 139     private CloudClientListener listener;
 140     private boolean remoteAccessEnabled;
 141     private Set<String> exposedItems;
 142
 143     /**
 144      * Back-off strategy for reconnecting when manual reconnection is needed
 145      */
 146     private final Backoff reconnectBackoff = new Backoff();
 147
 148     /*
 149      * Delay reconnect scheduler pool
 150      *
 151      */
 152     protected final ScheduledExecutorService scheduler = ThreadPoolManager
 153             .getScheduledPool(ThreadPoolManager.THREAD_POOL_NAME_COMMON);
 154
 155     @SuppressWarnings("null")
 156     private final AtomicReference<Optional<ScheduledFuture<?>>> reconnectFuture = new AtomicReference<>(
 157             Optional.empty());
 158
 159     /**
 160      * Constructor of CloudClient
 161      *
 162      * @param uuid openHAB's UUID to connect to the openHAB Cloud
 163      * @param secret openHAB's Secret to connect to the openHAB Cloud
 164      * @param remoteAccessEnabled Allow the openHAB Cloud to be used as a remote proxy
 165      * @param exposedItems Items that are made available to apps connected to the openHAB Cloud
 166      */
 167     public CloudClient(HttpClient httpClient, String uuid, String secret, String baseURL, String localBaseUrl,
 168             boolean remoteAccessEnabled, Set<String> exposedItems) {
 169         this.uuid = uuid;
 170         this.secret = secret;
 171         this.baseURL = baseURL;
 172         this.localBaseUrl = localBaseUrl;
 173         this.remoteAccessEnabled = remoteAccessEnabled;
 174         this.exposedItems = exposedItems;
 175         this.jettyClient = httpClient;
 176         reconnectBackoff.setMin(RECONNECT_MIN);
 177         reconnectBackoff.setMax(RECONNECT_MAX);
 178         reconnectBackoff.setJitter(RECONNECT_JITTER);
 179     }
 180
 181     /**
 182      * Connect to the openHAB Cloud
 183      */
 184
 185     public void connect() {
 186         try {
 187             Options options = new Options();
 188             options.transports = new String[] { WebSocket.NAME };
 189             options.reconnection = true;
 190             options.reconnectionAttempts = Integer.MAX_VALUE;
 191             options.reconnectionDelay = RECONNECT_MIN;
 192             options.reconnectionDelayMax = RECONNECT_MAX;
 193             options.randomizationFactor = RECONNECT_JITTER;
 194             options.timeout = READ_TIMEOUT;
 195             Builder okHttpBuilder = new Builder();
 196             okHttpBuilder.readTimeout(READ_TIMEOUT, TimeUnit.MILLISECONDS);
 197             if (logger.isTraceEnabled()) {
 198                 // When trace level logging is enabled, we activate further logging of HTTP calls
 199                 // of the Socket.IO library
 200                 HttpLoggingInterceptor loggingInterceptor = new HttpLoggingInterceptor();
 201                 loggingInterceptor.setLevel(Level.BASIC);
 202                 okHttpBuilder.addInterceptor(loggingInterceptor);
 203                 okHttpBuilder.addNetworkInterceptor(loggingInterceptor);
 204             }
 205             options.callFactory = okHttpBuilder.build();
 206             options.webSocketFactory = okHttpBuilder.build();
 207             socket = IO.socket(baseURL, options);
 208             URL parsed = new URL(baseURL);
 209             protocol = parsed.getProtocol();
 210         } catch (URISyntaxException e) {
 211             logger.error("Error creating Socket.IO: {}", e.getMessage());
 212             return;
 213         } catch (MalformedURLException e) {
 214             logger.error("Error parsing baseURL to get protocol, assuming https. Error: {}", e.getMessage());
 215             return;
 216         }
 217         //
 218         // socket manager events
 219         //
 220         socket.io()//
 221                 .on(Manager.EVENT_TRANSPORT, args -> {
 222                     logger.trace("Manager.EVENT_TRANSPORT");
 223                     Transport transport = (Transport) args[0];
 224                     transport.on(Transport.EVENT_REQUEST_HEADERS, new Emitter.Listener() {
 225                         @Override
 226                         public void call(Object... args) {
 227                             logger.trace("Transport.EVENT_REQUEST_HEADERS");
 228                             @SuppressWarnings("unchecked")
 229                             Map<String, List<String>> headers = (Map<String, List<String>>) args[0];
 230                             headers.put("uuid", List.of(uuid));
 231                             headers.put("secret", List.of(secret));
 232                             headers.put("openhabversion", List.of(OpenHAB.getVersion()));
 233                             headers.put("clientversion", List.of(CloudService.clientVersion));
 234                             headers.put("remoteaccess", List.of(((Boolean) remoteAccessEnabled).toString()));
 235                         }
 236                     });
 237                 })//
 238                 .on(Manager.EVENT_CONNECT_ERROR, args -> {
 239                     if (args.length > 0) {
 240                         if (args[0] instanceof Exception e) {
 241                             logger.debug(
 242                                     "Error connecting to the openHAB Cloud instance: {} {}. Should reconnect automatically.",
 243                                     e.getClass().getSimpleName(), e.getMessage());
 244                         } else {
 245                             logger.debug(
 246                                     "Error connecting to the openHAB Cloud instance: {}. Should reconnect automatically.",
 247                                     args[0]);
 248                         }
 249                     } else {
 250                         logger.debug("Error connecting to the openHAB Cloud instance. Should reconnect automatically.");
 251                     }
 252                 })//
 253                 .on(Manager.EVENT_OPEN, args -> logger.debug("Socket.IO OPEN"))//
 254                 .on(Manager.EVENT_CLOSE, args -> logger.debug("Socket.IO CLOSE: {}", args[0]))//
 255                 .on(Manager.EVENT_PACKET, args -> {
 256                     int packetTypeIndex = -1;
 257                     String type = "<unexpected packet type>";
 258                     if (args.length == 1 && args[0] instanceof Packet<?> packet) {
 259                         packetTypeIndex = packet.type;
 260
 261                         if (packetTypeIndex < Parser.types.length) {
 262                             type = Parser.types[packetTypeIndex];
 263                         } else {
 264                             type = "<unknown type>";
 265                         }
 266                     }
 267                     logger.trace("Socket.IO Packet: {} ({})", type, packetTypeIndex);
 268                 })//
 269         ;
 270
 271         //
 272         // socket events
 273         //
 274         socket.on(Socket.EVENT_CONNECT, args -> {
 275             logger.debug("Socket.IO connected");
 276             isConnected = true;
 277             onConnect();
 278         })//
 279                 .on(Socket.EVENT_CONNECTING, args -> logger.debug("Socket.IO connecting"))//
 280                 .on(Socket.EVENT_RECONNECTING, args -> logger.debug("Socket.IO re-connecting (attempt {})", args[0]))//
 281                 .on(Socket.EVENT_RECONNECT,
 282                         args -> logger.debug("Socket.IO re-connected successfully (attempt {})", args[0]))//
 283                 .on(Socket.EVENT_RECONNECT_ERROR, args -> {
 284                     if (args[0] instanceof Exception e) {
 285                         logger.debug("Socket.IO re-connect attempt error: {} {}", e.getClass().getSimpleName(),
 286                                 e.getMessage());
 287                     } else {
 288                         logger.debug("Socket.IO re-connect attempt error: {}", args[0]);
 289                     }
 290                 })//
 291                 .on(Socket.EVENT_RECONNECT_FAILED,
 292                         args -> logger.debug("Socket.IO re-connect attempts failed. Stopping reconnection."))//
 293                 .on(Socket.EVENT_DISCONNECT, args -> {
 294                     String message = args.length > 0 ? args[0].toString() : "";
 295                     logger.warn("Socket.IO disconnected: {}", message);
 296                     isConnected = false;
 297                     onDisconnect();
 298                     // https://github.com/socketio/socket.io-client/commit/afb952d854e1d8728ce07b7c3a9f0dee2a61ef4e
 299                     if ("io server disconnect".equals(message)) {
 300                         socket.close();
 301                         long delay = reconnectBackoff.duration();
 302                         logger.warn("Reconnecting after {} ms.", delay);
 303                         scheduleReconnect(delay);
 304                     }
 305                 })//
 306                 .on(Socket.EVENT_ERROR, args -> {
 307                     if (CloudClient.this.socket.connected()) {
 308                         if (args.length > 0) {
 309                             if (args[0] instanceof Exception e) {
 310                                 logger.warn("Error during communication: {} {}", e.getClass().getSimpleName(),
 311                                         e.getMessage());
 312                             } else {
 313                                 logger.warn("Error during communication: {}", args[0]);
 314                             }
 315                         } else {
 316                             logger.warn("Error during communication");
 317                         }
 318                     } else {
 319                         // We are not connected currently, manual reconnection is needed to keep trying to
 320                         // (re-)establish
 321                         // connection.
 322                         //
 323                         // Socket.IO 1.x java client: 'error' event is emitted from Socket on connection errors that
 324                         // are not
 325                         // retried, but also with error that are automatically retried. If we
 326                         //
 327                         // Note how this is different in Socket.IO 2.x java client, Socket emits 'connect_error'
 328                         // event.
 329                         // OBS: Don't get confused with Socket IO 2.x docs online, in 1.x connect_error is emitted
 330                         // also on
 331                         // errors that are retried by the library automatically!
 332                         long delay = reconnectBackoff.duration();
 333                         // Try reconnecting on connection errors
 334                         if (args.length > 0) {
 335                             if (args[0] instanceof Exception e) {
 336                                 logger.warn(
 337                                         "Error connecting to the openHAB Cloud instance: {} {}. Reconnecting after {} ms.",
 338                                         e.getClass().getSimpleName(), e.getMessage(), delay);
 339                             } else {
 340                                 logger.warn(
 341                                         "Error connecting to the openHAB Cloud instance: {}. Reconnecting after {} ms.",
 342                                         args[0], delay);
 343                             }
 344                         } else {
 345                             logger.warn("Error connecting to the openHAB Cloud instance. Reconnecting.");
 346                         }
 347                         socket.close();
 348                         scheduleReconnect(delay);
 349                     }
 350                 })//
 351
 352                 .on(Socket.EVENT_PING, args -> logger.debug("Socket.IO ping"))//
 353                 .on(Socket.EVENT_PONG, args -> logger.debug("Socket.IO pong: {} ms", args[0]))//
 354                 .on("request", args -> onEvent("request", (JSONObject) args[0]))//
 355                 .on("cancel", args -> onEvent("cancel", (JSONObject) args[0]))//
 356                 .on("command", args -> onEvent("command", (JSONObject) args[0]))//
 357         ;
 358         socket.connect();
 359     }
 360
 361     /**
 362      * Callback method for socket.io client which is called when connection is established
 363      */
 364
 365     public void onConnect() {
 366         logger.info("Connected to the openHAB Cloud service (UUID = {}, base URL = {})", censored(this.uuid),
 367                 this.localBaseUrl);
 368         reconnectBackoff.reset();
 369         isConnected = true;
 370     }
 371
 372     /**
 373      * Callback method for socket.io client which is called when disconnect occurs
 374      */
 375
 376     public void onDisconnect() {
 377         logger.info("Disconnected from the openHAB Cloud service (UUID = {}, base URL = {})", censored(this.uuid),
 378                 this.localBaseUrl);
 379         isConnected = false;
 380         // And clean up the list of running requests
 381         runningRequests.clear();
 382     }
 383
 384     /**
 385      * Callback method for socket.io client which is called when a message is received
 386      */
 387
 388     public void onEvent(String event, JSONObject data) {
 389         logger.debug("on(): {}", event);
 390         if ("command".equals(event)) {
 391             handleCommandEvent(data);
 392             return;
 393         }
 394         if (remoteAccessEnabled) {
 395             if ("request".equals(event)) {
 396                 handleRequestEvent(data);
 397             } else if ("cancel".equals(event)) {
 398                 handleCancelEvent(data);
 399             } else {
 400                 logger.warn("Unsupported event from openHAB Cloud: {}", event);
 401             }
 402         }
 403     }
 404
 405     private void handleRequestEvent(JSONObject data) {
 406         try {
 407             // Get unique request Id
 408             int requestId = data.getInt("id");
 409             logger.debug("Got request {}", requestId);
 410             // Get request path
 411             String requestPath = data.getString("path");
 412             logger.debug("Path {}", requestPath);
 413             // Get request method
 414             String requestMethod = data.getString("method");
 415             logger.debug("Method {}", requestMethod);
 416             // Get JSONObject for request headers
 417             JSONObject requestHeadersJson = data.getJSONObject("headers");
 418             logger.debug("Headers: {}", requestHeadersJson.toString());
 419             // Get request body
 420             String requestBody = data.getString("body");
 421             logger.trace("Body {}", requestBody);
 422             // Get JSONObject for request query parameters
 423             JSONObject requestQueryJson = data.getJSONObject("query");
 424             logger.debug("Query {}", requestQueryJson.toString());
 425             // Create URI builder with base request URI of openHAB and path from request
 426             String newPath = URIUtil.addPaths(localBaseUrl, requestPath);
 427             Iterator<String> queryIterator = requestQueryJson.keys();
 428             // Add query parameters to URI builder, if any
 429             newPath += "?";
 430             while (queryIterator.hasNext()) {
 431                 String queryName = queryIterator.next();
 432                 newPath += queryName;
 433                 newPath += "=";
 434                 newPath += URLEncoder.encode(requestQueryJson.getString(queryName), "UTF-8");
 435                 if (queryIterator.hasNext()) {
 436                     newPath += "&";
 437                 }
 438             }
 439             // Finally get the future request URI
 440             URI requestUri = new URI(newPath);
 441             // All preparations which are common for different methods are done
 442             // Now perform the request to openHAB
 443             // If method is GET
 444             logger.debug("Request method is {}", requestMethod);
 445             Request request = jettyClient.newRequest(requestUri);
 446             setRequestHeaders(request, requestHeadersJson);
 447             String proto = protocol;
 448             if (data.has("protocol")) {
 449                 proto = data.getString("protocol");
 450             }
 451             request.header("X-Forwarded-Proto", proto);
 452             HttpMethod method = HttpMethod.fromString(requestMethod);
 453             if (method == null) {
 454                 logger.debug("Unsupported request method {}", requestMethod);
 455                 return;
 456             }
 457             request.method(method);
 458             if (!requestBody.isEmpty()) {
 459                 request.content(new BytesContentProvider(requestBody.getBytes()));
 460             }
 461
 462             request.onResponseHeaders(response -> {
 463                 logger.debug("onHeaders {}", requestId);
 464                 JSONObject responseJson = new JSONObject();
 465                 try {
 466                     responseJson.put("id", requestId);
 467                     responseJson.put("headers", getJSONHeaders(response.getHeaders()));
 468                     responseJson.put("responseStatusCode", response.getStatus());
 469                     responseJson.put("responseStatusText", "OK");
 470                     socket.emit("responseHeader", responseJson);
 471                     logger.trace("Sent headers to request {}", requestId);
 472                     logger.trace("{}", responseJson.toString());
 473                 } catch (JSONException e) {
 474                     logger.debug("{}", e.getMessage());
 475                 }
 476             }).onResponseContent((theResponse, content) -> {
 477                 logger.debug("onResponseContent: {}, content size {}", requestId, String.valueOf(content.remaining()));
 478                 JSONObject responseJson = new JSONObject();
 479                 try {
 480                     responseJson.put("id", requestId);
 481                     responseJson.put("body", BufferUtil.toArray(content));
 482                     if (logger.isTraceEnabled()) {
 483                         logger.trace("{}", StandardCharsets.UTF_8.decode(content).toString());
 484                     }
 485                     socket.emit("responseContentBinary", responseJson);
 486                     logger.trace("Sent content to request {}", requestId);
 487                 } catch (JSONException e) {
 488                     logger.debug("{}", e.getMessage());
 489                 }
 490             }).onRequestFailure((origRequest, failure) -> {
 491                 logger.debug("onRequestFailure: {},  {}", requestId, failure.getMessage());
 492                 JSONObject responseJson = new JSONObject();
 493                 try {
 494                     responseJson.put("id", requestId);
 495                     responseJson.put("responseStatusText", "openHAB connection error: " + failure.getMessage());
 496                     socket.emit("responseError", responseJson);
 497                 } catch (JSONException e) {
 498                     logger.debug("{}", e.getMessage());
 499                 }
 500             }).send(result -> {
 501                 logger.debug("onComplete: {}", requestId);
 502                 // Remove this request from list of running requests
 503                 runningRequests.remove(requestId);
 504                 if ((result != null && result.isFailed())
 505                         && (result.getResponse() != null && result.getResponse().getStatus() != HttpStatus.OK_200)) {
 506                     if (result.getFailure() != null) {
 507                         logger.debug("Jetty request {} failed: {}", requestId, result.getFailure().getMessage());
 508                     }
 509                     if (result.getRequestFailure() != null) {
 510                         logger.debug("Request Failure: {}", result.getRequestFailure().getMessage());
 511                     }
 512                     if (result.getResponseFailure() != null) {
 513                         logger.debug("Response Failure: {}", result.getResponseFailure().getMessage());
 514                     }
 515                 }
 516                 JSONObject responseJson = new JSONObject();
 517                 try {
 518                     responseJson.put("id", requestId);
 519                     socket.emit("responseFinished", responseJson);
 520                     logger.debug("Finished responding to request {}", requestId);
 521                 } catch (JSONException e) {
 522                     logger.debug("{}", e.getMessage());
 523                 }
 524             });
 525
 526             // If successfully submitted request to http client, add it to the list of currently
 527             // running requests to be able to cancel it if needed
 528             runningRequests.put(requestId, request);
 529         } catch (JSONException | IOException | URISyntaxException e) {
 530             logger.debug("{}", e.getMessage());
 531         }
 532     }
 533
 534     private void setRequestHeaders(Request request, JSONObject requestHeadersJson) {
 535         Iterator<String> headersIterator = requestHeadersJson.keys();
 536         // Convert JSONObject of headers into Header ArrayList
 537         while (headersIterator.hasNext()) {
 538             String headerName = headersIterator.next();
 539             String headerValue;
 540             try {
 541                 headerValue = requestHeadersJson.getString(headerName);
 542                 logger.debug("Jetty set header {} = {}", headerName, headerValue);
 543                 if (!headerName.equalsIgnoreCase("Content-Length")) {
 544                     request.header(headerName, headerValue);
 545                 }
 546             } catch (JSONException e) {
 547                 logger.warn("Error processing request headers: {}", e.getMessage());
 548             }
 549         }
 550     }
 551
 552     private void handleCancelEvent(JSONObject data) {
 553         try {
 554             int requestId = data.getInt("id");
 555             logger.debug("Received cancel for request {}", requestId);
 556             // Find and abort running request
 557             Request request = runningRequests.get(requestId);
 558             if (request != null) {
 559                 request.abort(new InterruptedException());
 560                 runningRequests.remove(requestId);
 561             }
 562         } catch (JSONException e) {
 563             logger.debug("{}", e.getMessage());
 564         }
 565     }
 566
 567     private void handleCommandEvent(JSONObject data) {
 568         String itemName = data.getString("item");
 569         if (exposedItems.contains(itemName)) {
 570             try {
 571                 logger.debug("Received command {} for item {}.", data.getString("command"), itemName);
 572                 if (this.listener != null) {
 573                     this.listener.sendCommand(itemName, data.getString("command"));
 574                 }
 575             } catch (JSONException e) {
 576                 logger.debug("{}", e.getMessage());
 577             }
 578         } else {
 579             logger.warn("Received command from openHAB Cloud for item '{}', which is not exposed.", itemName);
 580         }
 581     }
 582
 583     /**
 584      * This method sends notification to the openHAB Cloud
 585      *
 586      * @param userId openHAB Cloud user id
 587      * @param message notification message text
 588      * @param icon name of the icon for this notification
 589      * @param severity severity name for this notification
 590      */
 591     public void sendNotification(String userId, String message, @Nullable String icon, @Nullable String severity) {
 592         if (isConnected()) {
 593             JSONObject notificationMessage = new JSONObject();
 594             try {
 595                 notificationMessage.put("userId", userId);
 596                 notificationMessage.put("message", message);
 597                 notificationMessage.put("icon", icon);
 598                 notificationMessage.put("severity", severity);
 599                 socket.emit("notification", notificationMessage);
 600             } catch (JSONException e) {
 601                 logger.debug("{}", e.getMessage());
 602             }
 603         } else {
 604             logger.debug("No connection, notification is not sent");
 605         }
 606     }
 607
 608     /**
 609      * This method sends log notification to the openHAB Cloud
 610      *
 611      * @param message notification message text
 612      * @param icon name of the icon for this notification
 613      * @param severity severity name for this notification
 614      */
 615     public void sendLogNotification(String message, @Nullable String icon, @Nullable String severity) {
 616         if (isConnected()) {
 617             JSONObject notificationMessage = new JSONObject();
 618             try {
 619                 notificationMessage.put("message", message);
 620                 notificationMessage.put("icon", icon);
 621                 notificationMessage.put("severity", severity);
 622                 socket.emit("lognotification", notificationMessage);
 623             } catch (JSONException e) {
 624                 logger.debug("{}", e.getMessage());
 625             }
 626         } else {
 627             logger.debug("No connection, notification is not sent");
 628         }
 629     }
 630
 631     /**
 632      * This method sends broadcast notification to the openHAB Cloud
 633      *
 634      * @param message notification message text
 635      * @param icon name of the icon for this notification
 636      * @param severity severity name for this notification
 637      */
 638     public void sendBroadcastNotification(String message, @Nullable String icon, @Nullable String severity) {
 639         if (isConnected()) {
 640             JSONObject notificationMessage = new JSONObject();
 641             try {
 642                 notificationMessage.put("message", message);
 643                 notificationMessage.put("icon", icon);
 644                 notificationMessage.put("severity", severity);
 645                 socket.emit("broadcastnotification", notificationMessage);
 646             } catch (JSONException e) {
 647                 logger.debug("{}", e.getMessage());
 648             }
 649         } else {
 650             logger.debug("No connection, notification is not sent");
 651         }
 652     }
 653
 654     /**
 655      * Send item update to openHAB Cloud
 656      *
 657      * @param itemName the name of the item
 658      * @param itemState updated item state
 659      *
 660      */
 661     public void sendItemUpdate(String itemName, String itemState) {
 662         if (isConnected()) {
 663             logger.debug("Sending update '{}' for item '{}'", itemState, itemName);
 664             JSONObject itemUpdateMessage = new JSONObject();
 665             try {
 666                 itemUpdateMessage.put("itemName", itemName);
 667                 itemUpdateMessage.put("itemStatus", itemState);
 668                 socket.emit("itemupdate", itemUpdateMessage);
 669             } catch (JSONException e) {
 670                 logger.debug("{}", e.getMessage());
 671             }
 672         } else {
 673             logger.debug("No connection, Item update is not sent");
 674         }
 675     }
 676
 677     /**
 678      * Returns true if openHAB Cloud connection is active
 679      */
 680     public boolean isConnected() {
 681         return isConnected;
 682     }
 683
 684     /**
 685      * Disconnect from openHAB Cloud
 686      */
 687     public void shutdown() {
 688         logger.info("Shutting down openHAB Cloud service connection");
 689         reconnectFuture.get().ifPresent(future -> future.cancel(true));
 690         socket.disconnect();
 691     }
 692
 693     public void setListener(CloudClientListener listener) {
 694         this.listener = listener;
 695     }
 696
 697     private void scheduleReconnect(long delay) {
 698         reconnectFuture.getAndSet(Optional.of(scheduler.schedule(new Runnable() {
 699             @Override
 700             public void run() {
 701                 socket.connect();
 702             }
 703         }, delay, TimeUnit.MILLISECONDS))).ifPresent(future -> future.cancel(true));
 704     }
 705
 706     private JSONObject getJSONHeaders(HttpFields httpFields) {
 707         JSONObject headersJSON = new JSONObject();
 708         try {
 709             for (HttpField field : httpFields) {
 710                 headersJSON.put(field.getName(), field.getValue());
 711             }
 712         } catch (JSONException e) {
 713             logger.warn("Error forming response headers: {}", e.getMessage());
 714         }
 715         return headersJSON;
 716     }
 717
 718     private static String censored(String secret) {
 719         if (secret.length() < 4) {
 720             return "*******";
 721         }
 722         return secret.substring(0, 2) + "..." + secret.substring(secret.length() - 2, secret.length());
 723     }
 724 }