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

   1 /**
   2  * Copyright (c) 2010-2021 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.ByteBuffer;
  22 import java.util.Iterator;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Set;
  26 import java.util.concurrent.ConcurrentHashMap;
  27 import java.util.concurrent.TimeUnit;
  28
  29 import org.eclipse.jdt.annotation.Nullable;
  30 import org.eclipse.jetty.client.HttpClient;
  31 import org.eclipse.jetty.client.api.Request;
  32 import org.eclipse.jetty.client.api.Request.FailureListener;
  33 import org.eclipse.jetty.client.api.Response;
  34 import org.eclipse.jetty.client.api.Response.ContentListener;
  35 import org.eclipse.jetty.client.api.Response.HeadersListener;
  36 import org.eclipse.jetty.client.api.Result;
  37 import org.eclipse.jetty.client.util.BytesContentProvider;
  38 import org.eclipse.jetty.http.HttpField;
  39 import org.eclipse.jetty.http.HttpFields;
  40 import org.eclipse.jetty.http.HttpMethod;
  41 import org.eclipse.jetty.http.HttpStatus;
  42 import org.eclipse.jetty.util.BufferUtil;
  43 import org.eclipse.jetty.util.URIUtil;
  44 import org.json.JSONException;
  45 import org.json.JSONObject;
  46 import org.openhab.core.OpenHAB;
  47 import org.openhab.core.common.ThreadPoolManager;
  48 import org.slf4j.Logger;
  49 import org.slf4j.LoggerFactory;
  50
  51 import io.socket.client.IO;
  52 import io.socket.client.Manager;
  53 import io.socket.client.Socket;
  54 import io.socket.emitter.Emitter;
  55 import io.socket.engineio.client.Transport;
  56
  57 /**
  58  * This class provides communication between openHAB and the openHAB Cloud service.
  59  * It also implements async http proxy for serving requests from user to
  60  * openHAB through the openHAB Cloud. It uses Socket.IO connection to connect to
  61  * the openHAB Cloud service and Jetty Http client to send local http requests to
  62  * openHAB.
  63  *
  64  * @author Victor Belov - Initial contribution
  65  * @author Kai Kreuzer - migrated code to new Jetty client and ESH APIs
  66  */
  67 public class CloudClient {
  68     /*
  69      * Logger for this class
  70      */
  71     private final Logger logger = LoggerFactory.getLogger(CloudClient.class);
  72
  73     /*
  74      * This variable holds base URL for the openHAB Cloud connections
  75      */
  76     private final String baseURL;
  77
  78     /*
  79      * This variable holds openHAB's UUID for authenticating and connecting to the openHAB Cloud
  80      */
  81     private final String uuid;
  82
  83     /*
  84      * This variable holds openHAB's secret for authenticating and connecting to the openHAB Cloud
  85      */
  86     private final String secret;
  87
  88     /*
  89      * This variable holds local openHAB's base URL for connecting to the local openHAB instance
  90      */
  91     private final String localBaseUrl;
  92
  93     /*
  94      * This variable holds instance of Jetty HTTP client to make requests to local openHAB
  95      */
  96     private final HttpClient jettyClient;
  97
  98     /*
  99      * This map holds HTTP requests to local openHAB which are currently running
 100      */
 101     private final Map<Integer, Request> runningRequests = new ConcurrentHashMap<>();
 102
 103     /*
 104      * This variable indicates if connection to the openHAB Cloud is currently in an established state
 105      */
 106     private boolean isConnected;
 107
 108     /*
 109      * This variable holds version of local openHAB
 110      */
 111     private String openHABVersion;
 112
 113     /*
 114      * This variable holds instance of Socket.IO client class which provides communication
 115      * with the openHAB Cloud
 116      */
 117     private Socket socket;
 118
 119     /*
 120      * The protocol of the openHAB-cloud URL.
 121      */
 122     private String protocol = "https";
 123
 124     /*
 125      * This variable holds instance of CloudClientListener which provides callbacks to communicate
 126      * certain events from the openHAB Cloud back to openHAB
 127      */
 128     private CloudClientListener listener;
 129     private boolean remoteAccessEnabled;
 130     private Set<String> exposedItems;
 131
 132     /**
 133      * Constructor of CloudClient
 134      *
 135      * @param uuid openHAB's UUID to connect to the openHAB Cloud
 136      * @param secret openHAB's Secret to connect to the openHAB Cloud
 137      * @param remoteAccessEnabled Allow the openHAB Cloud to be used as a remote proxy
 138      * @param exposedItems Items that are made available to apps connected to the openHAB Cloud
 139      */
 140     public CloudClient(HttpClient httpClient, String uuid, String secret, String baseURL, String localBaseUrl,
 141             boolean remoteAccessEnabled, Set<String> exposedItems) {
 142         this.uuid = uuid;
 143         this.secret = secret;
 144         this.baseURL = baseURL;
 145         this.localBaseUrl = localBaseUrl;
 146         this.remoteAccessEnabled = remoteAccessEnabled;
 147         this.exposedItems = exposedItems;
 148         this.jettyClient = httpClient;
 149     }
 150
 151     /**
 152      * Connect to the openHAB Cloud
 153      */
 154
 155     public void connect() {
 156         try {
 157             socket = IO.socket(baseURL);
 158             URL parsed = new URL(baseURL);
 159             protocol = parsed.getProtocol();
 160         } catch (URISyntaxException e) {
 161             logger.error("Error creating Socket.IO: {}", e.getMessage());
 162         } catch (MalformedURLException e) {
 163             logger.error("Error parsing baseURL to get protocol, assuming https. Error: {}", e.getMessage());
 164         }
 165         socket.io().on(Manager.EVENT_TRANSPORT, new Emitter.Listener() {
 166             @Override
 167             public void call(Object... args) {
 168                 logger.trace("Manager.EVENT_TRANSPORT");
 169                 Transport transport = (Transport) args[0];
 170                 transport.on(Transport.EVENT_REQUEST_HEADERS, new Emitter.Listener() {
 171                     @Override
 172                     public void call(Object... args) {
 173                         logger.trace("Transport.EVENT_REQUEST_HEADERS");
 174                         @SuppressWarnings("unchecked")
 175                         Map<String, List<String>> headers = (Map<String, List<String>>) args[0];
 176                         headers.put("uuid", List.of(uuid));
 177                         headers.put("secret", List.of(secret));
 178                         headers.put("openhabversion", List.of(OpenHAB.getVersion()));
 179                         headers.put("clientversion", List.of(CloudService.clientVersion));
 180                         headers.put("remoteaccess", List.of(((Boolean) remoteAccessEnabled).toString()));
 181                     }
 182                 });
 183             }
 184         });
 185         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
 186             @Override
 187             public void call(Object... args) {
 188                 logger.debug("Socket.IO connected");
 189                 isConnected = true;
 190                 onConnect();
 191             }
 192         }).on(Socket.EVENT_DISCONNECT, new Emitter.Listener() {
 193             @Override
 194             public void call(Object... args) {
 195                 logger.debug("Socket.IO disconnected");
 196                 isConnected = false;
 197                 onDisconnect();
 198             }
 199         }).on(Socket.EVENT_ERROR, new Emitter.Listener() {
 200             @Override
 201             public void call(Object... args) {
 202                 if (logger.isDebugEnabled()) {
 203                     logger.error("Error connecting to the openHAB Cloud instance: {}", args[0]);
 204                 } else {
 205                     logger.error("Error connecting to the openHAB Cloud instance");
 206                 }
 207             }
 208         }).on("request", new Emitter.Listener() {
 209             @Override
 210             public void call(Object... args) {
 211                 onEvent("request", (JSONObject) args[0]);
 212             }
 213         }).on("cancel", new Emitter.Listener() {
 214             @Override
 215             public void call(Object... args) {
 216                 onEvent("cancel", (JSONObject) args[0]);
 217             }
 218         }).on("command", new Emitter.Listener() {
 219
 220             @Override
 221             public void call(Object... args) {
 222                 onEvent("command", (JSONObject) args[0]);
 223             }
 224         });
 225         socket.connect();
 226     }
 227
 228     /**
 229      * Callback method for socket.io client which is called when connection is established
 230      */
 231
 232     public void onConnect() {
 233         logger.info("Connected to the openHAB Cloud service (UUID = {}, base URL = {})", this.uuid, this.localBaseUrl);
 234         isConnected = true;
 235     }
 236
 237     /**
 238      * Callback method for socket.io client which is called when disconnect occurs
 239      */
 240
 241     public void onDisconnect() {
 242         logger.info("Disconnected from the openHAB Cloud service (UUID = {}, base URL = {})", this.uuid,
 243                 this.localBaseUrl);
 244         isConnected = false;
 245         // And clean up the list of running requests
 246         runningRequests.clear();
 247     }
 248
 249     /**
 250      * Callback method for socket.io client which is called when an error occurs
 251      */
 252
 253     public void onError(IOException error) {
 254         logger.debug("{}", error.getMessage());
 255     }
 256
 257     /**
 258      * Callback method for socket.io client which is called when a message is received
 259      */
 260
 261     public void onEvent(String event, JSONObject data) {
 262         logger.debug("on(): {}", event);
 263         if ("command".equals(event)) {
 264             handleCommandEvent(data);
 265             return;
 266         }
 267         if (remoteAccessEnabled) {
 268             if ("request".equals(event)) {
 269                 handleRequestEvent(data);
 270             } else if ("cancel".equals(event)) {
 271                 handleCancelEvent(data);
 272             } else {
 273                 logger.warn("Unsupported event from openHAB Cloud: {}", event);
 274             }
 275         }
 276     }
 277
 278     private void handleRequestEvent(JSONObject data) {
 279         try {
 280             // Get unique request Id
 281             int requestId = data.getInt("id");
 282             logger.debug("Got request {}", requestId);
 283             // Get request path
 284             String requestPath = data.getString("path");
 285             // Get request method
 286             String requestMethod = data.getString("method");
 287             // Get request body
 288             String requestBody = data.getString("body");
 289             // Get JSONObject for request headers
 290             JSONObject requestHeadersJson = data.getJSONObject("headers");
 291             logger.debug("{}", requestHeadersJson.toString());
 292             // Get JSONObject for request query parameters
 293             JSONObject requestQueryJson = data.getJSONObject("query");
 294             // Create URI builder with base request URI of openHAB and path from request
 295             String newPath = URIUtil.addPaths(localBaseUrl, requestPath);
 296             Iterator<String> queryIterator = requestQueryJson.keys();
 297             // Add query parameters to URI builder, if any
 298             newPath += "?";
 299             while (queryIterator.hasNext()) {
 300                 String queryName = queryIterator.next();
 301                 newPath += queryName;
 302                 newPath += "=";
 303                 newPath += URLEncoder.encode(requestQueryJson.getString(queryName), "UTF-8");
 304                 if (queryIterator.hasNext()) {
 305                     newPath += "&";
 306                 }
 307             }
 308             // Finally get the future request URI
 309             URI requestUri = new URI(newPath);
 310             // All preparations which are common for different methods are done
 311             // Now perform the request to openHAB
 312             // If method is GET
 313             logger.debug("Request method is {}", requestMethod);
 314             Request request = jettyClient.newRequest(requestUri);
 315             setRequestHeaders(request, requestHeadersJson);
 316             String proto = protocol;
 317             if (data.has("protocol")) {
 318                 proto = data.getString("protocol");
 319             }
 320             request.header("X-Forwarded-Proto", proto);
 321
 322             if (requestMethod.equals("GET")) {
 323                 request.method(HttpMethod.GET);
 324             } else if (requestMethod.equals("POST")) {
 325                 request.method(HttpMethod.POST);
 326                 request.content(new BytesContentProvider(requestBody.getBytes()));
 327             } else if (requestMethod.equals("PUT")) {
 328                 request.method(HttpMethod.PUT);
 329                 request.content(new BytesContentProvider(requestBody.getBytes()));
 330             } else {
 331                 // TODO: Reject unsupported methods
 332                 logger.warn("Unsupported request method {}", requestMethod);
 333                 return;
 334             }
 335             ResponseListener listener = new ResponseListener(requestId);
 336             request.onResponseHeaders(listener).onResponseContent(listener).onRequestFailure(listener).send(listener);
 337             // If successfully submitted request to http client, add it to the list of currently
 338             // running requests to be able to cancel it if needed
 339             runningRequests.put(requestId, request);
 340         } catch (JSONException | IOException | URISyntaxException e) {
 341             logger.debug("{}", e.getMessage());
 342         }
 343     }
 344
 345     private void setRequestHeaders(Request request, JSONObject requestHeadersJson) {
 346         Iterator<String> headersIterator = requestHeadersJson.keys();
 347         // Convert JSONObject of headers into Header ArrayList
 348         while (headersIterator.hasNext()) {
 349             String headerName = headersIterator.next();
 350             String headerValue;
 351             try {
 352                 headerValue = requestHeadersJson.getString(headerName);
 353                 logger.debug("Jetty set header {} = {}", headerName, headerValue);
 354                 if (!headerName.equalsIgnoreCase("Content-Length")) {
 355                     request.header(headerName, headerValue);
 356                 }
 357             } catch (JSONException e) {
 358                 logger.warn("Error processing request headers: {}", e.getMessage());
 359             }
 360         }
 361     }
 362
 363     private void handleCancelEvent(JSONObject data) {
 364         try {
 365             int requestId = data.getInt("id");
 366             logger.debug("Received cancel for request {}", requestId);
 367             // Find and abort running request
 368             Request request = runningRequests.get(requestId);
 369             if (request != null) {
 370                 request.abort(new InterruptedException());
 371                 runningRequests.remove(requestId);
 372             }
 373         } catch (JSONException e) {
 374             logger.debug("{}", e.getMessage());
 375         }
 376     }
 377
 378     private void handleCommandEvent(JSONObject data) {
 379         String itemName = data.getString("item");
 380         if (exposedItems.contains(itemName)) {
 381             try {
 382                 logger.debug("Received command {} for item {}.", data.getString("command"), itemName);
 383                 if (this.listener != null) {
 384                     this.listener.sendCommand(itemName, data.getString("command"));
 385                 }
 386             } catch (JSONException e) {
 387                 logger.debug("{}", e.getMessage());
 388             }
 389         } else {
 390             logger.warn("Received command from openHAB Cloud for item '{}', which is not exposed.", itemName);
 391         }
 392     }
 393
 394     /**
 395      * This method sends notification to the openHAB Cloud
 396      *
 397      * @param userId openHAB Cloud user id
 398      * @param message notification message text
 399      * @param icon name of the icon for this notification
 400      * @param severity severity name for this notification
 401      */
 402     public void sendNotification(String userId, String message, @Nullable String icon, @Nullable String severity) {
 403         if (isConnected()) {
 404             JSONObject notificationMessage = new JSONObject();
 405             try {
 406                 notificationMessage.put("userId", userId);
 407                 notificationMessage.put("message", message);
 408                 notificationMessage.put("icon", icon);
 409                 notificationMessage.put("severity", severity);
 410                 socket.emit("notification", notificationMessage);
 411             } catch (JSONException e) {
 412                 logger.debug("{}", e.getMessage());
 413             }
 414         } else {
 415             logger.debug("No connection, notification is not sent");
 416         }
 417     }
 418
 419     /**
 420      * This method sends log notification to the openHAB Cloud
 421      *
 422      * @param message notification message text
 423      * @param icon name of the icon for this notification
 424      * @param severity severity name for this notification
 425      */
 426     public void sendLogNotification(String message, @Nullable String icon, @Nullable String severity) {
 427         if (isConnected()) {
 428             JSONObject notificationMessage = new JSONObject();
 429             try {
 430                 notificationMessage.put("message", message);
 431                 notificationMessage.put("icon", icon);
 432                 notificationMessage.put("severity", severity);
 433                 socket.emit("lognotification", notificationMessage);
 434             } catch (JSONException e) {
 435                 logger.debug("{}", e.getMessage());
 436             }
 437         } else {
 438             logger.debug("No connection, notification is not sent");
 439         }
 440     }
 441
 442     /**
 443      * This method sends broadcast notification to the openHAB Cloud
 444      *
 445      * @param message notification message text
 446      * @param icon name of the icon for this notification
 447      * @param severity severity name for this notification
 448      */
 449     public void sendBroadcastNotification(String message, @Nullable String icon, @Nullable String severity) {
 450         if (isConnected()) {
 451             JSONObject notificationMessage = new JSONObject();
 452             try {
 453                 notificationMessage.put("message", message);
 454                 notificationMessage.put("icon", icon);
 455                 notificationMessage.put("severity", severity);
 456                 socket.emit("broadcastnotification", notificationMessage);
 457             } catch (JSONException e) {
 458                 logger.debug("{}", e.getMessage());
 459             }
 460         } else {
 461             logger.debug("No connection, notification is not sent");
 462         }
 463     }
 464
 465     /**
 466      * Send item update to openHAB Cloud
 467      *
 468      * @param itemName the name of the item
 469      * @param itemState updated item state
 470      *
 471      */
 472     public void sendItemUpdate(String itemName, String itemState) {
 473         if (isConnected()) {
 474             logger.debug("Sending update '{}' for item '{}'", itemState, itemName);
 475             JSONObject itemUpdateMessage = new JSONObject();
 476             try {
 477                 itemUpdateMessage.put("itemName", itemName);
 478                 itemUpdateMessage.put("itemStatus", itemState);
 479                 socket.emit("itemupdate", itemUpdateMessage);
 480             } catch (JSONException e) {
 481                 logger.debug("{}", e.getMessage());
 482             }
 483         } else {
 484             logger.debug("No connection, Item update is not sent");
 485         }
 486     }
 487
 488     /**
 489      * Returns true if openHAB Cloud connection is active
 490      */
 491     public boolean isConnected() {
 492         return isConnected;
 493     }
 494
 495     /**
 496      * Disconnect from openHAB Cloud
 497      */
 498     public void shutdown() {
 499         logger.info("Shutting down openHAB Cloud service connection");
 500         socket.disconnect();
 501     }
 502
 503     public String getOpenHABVersion() {
 504         return openHABVersion;
 505     }
 506
 507     public void setOpenHABVersion(String openHABVersion) {
 508         this.openHABVersion = openHABVersion;
 509     }
 510
 511     public void setListener(CloudClientListener listener) {
 512         this.listener = listener;
 513     }
 514
 515     /*
 516      * An internal class which forwards response headers and data back to the openHAB Cloud
 517      */
 518     private class ResponseListener
 519             implements Response.CompleteListener, HeadersListener, ContentListener, FailureListener {
 520
 521         private static final String THREADPOOL_OPENHABCLOUD = "openhabcloud";
 522         private int mRequestId;
 523         private boolean mHeadersSent = false;
 524
 525         public ResponseListener(int requestId) {
 526             mRequestId = requestId;
 527         }
 528
 529         private JSONObject getJSONHeaders(HttpFields httpFields) {
 530             JSONObject headersJSON = new JSONObject();
 531             try {
 532                 for (HttpField field : httpFields) {
 533                     headersJSON.put(field.getName(), field.getValue());
 534                 }
 535             } catch (JSONException e) {
 536                 logger.warn("Error forming response headers: {}", e.getMessage());
 537             }
 538             return headersJSON;
 539         }
 540
 541         @Override
 542         public void onComplete(Result result) {
 543             // Remove this request from list of running requests
 544             runningRequests.remove(mRequestId);
 545             if ((result != null && result.isFailed())
 546                     && (result.getResponse() != null && result.getResponse().getStatus() != HttpStatus.OK_200)) {
 547                 if (result.getFailure() != null) {
 548                     logger.warn("Jetty request {} failed: {}", mRequestId, result.getFailure().getMessage());
 549                 }
 550                 if (result.getRequestFailure() != null) {
 551                     logger.warn("Request Failure: {}", result.getRequestFailure().getMessage());
 552                 }
 553                 if (result.getResponseFailure() != null) {
 554                     logger.warn("Response Failure: {}", result.getResponseFailure().getMessage());
 555                 }
 556             }
 557
 558             /**
 559              * What is this? In some cases where latency is very low the myopenhab service
 560              * can receive responseFinished before the headers or content are received and I
 561              * cannot find another workaround to prevent it.
 562              */
 563             ThreadPoolManager.getScheduledPool(THREADPOOL_OPENHABCLOUD).schedule(() -> {
 564                 JSONObject responseJson = new JSONObject();
 565                 try {
 566                     responseJson.put("id", mRequestId);
 567                     socket.emit("responseFinished", responseJson);
 568                     logger.debug("Finished responding to request {}", mRequestId);
 569                 } catch (JSONException e) {
 570                     logger.debug("{}", e.getMessage());
 571                 }
 572             }, 1, TimeUnit.MILLISECONDS);
 573         }
 574
 575         @Override
 576         public synchronized void onFailure(Request request, Throwable failure) {
 577             JSONObject responseJson = new JSONObject();
 578             try {
 579                 responseJson.put("id", mRequestId);
 580                 responseJson.put("responseStatusText", "openHAB connection error: " + failure.getMessage());
 581                 socket.emit("responseError", responseJson);
 582             } catch (JSONException e) {
 583                 logger.debug("{}", e.getMessage());
 584             }
 585         }
 586
 587         @Override
 588         public void onContent(Response response, ByteBuffer content) {
 589             logger.debug("Jetty received response content of size {}", String.valueOf(content.remaining()));
 590             JSONObject responseJson = new JSONObject();
 591             try {
 592                 responseJson.put("id", mRequestId);
 593                 responseJson.put("body", BufferUtil.toArray(content));
 594                 socket.emit("responseContentBinary", responseJson);
 595                 logger.debug("Sent content to request {}", mRequestId);
 596             } catch (JSONException e) {
 597                 logger.debug("{}", e.getMessage());
 598             }
 599         }
 600
 601         @Override
 602         public void onHeaders(Response response) {
 603             if (!mHeadersSent) {
 604                 logger.debug("Jetty finished receiving response header");
 605                 JSONObject responseJson = new JSONObject();
 606                 mHeadersSent = true;
 607                 try {
 608                     responseJson.put("id", mRequestId);
 609                     responseJson.put("headers", getJSONHeaders(response.getHeaders()));
 610                     responseJson.put("responseStatusCode", response.getStatus());
 611                     responseJson.put("responseStatusText", "OK");
 612                     socket.emit("responseHeader", responseJson);
 613                     logger.debug("Sent headers to request {}", mRequestId);
 614                     logger.debug("{}", responseJson.toString());
 615                 } catch (JSONException e) {
 616                     logger.debug("{}", e.getMessage());
 617                 }
 618             }
 619         }
 620     }
 621 }