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

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