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.binding.miele.internal.handler;
  14
  15 import static org.openhab.binding.miele.internal.MieleBindingConstants.*;
  16
  17 import java.io.BufferedInputStream;
  18 import java.io.ByteArrayOutputStream;
  19 import java.io.IOException;
  20 import java.io.InputStream;
  21 import java.io.OutputStream;
  22 import java.io.StringReader;
  23 import java.net.DatagramPacket;
  24 import java.net.HttpURLConnection;
  25 import java.net.InetAddress;
  26 import java.net.MalformedURLException;
  27 import java.net.MulticastSocket;
  28 import java.net.SocketTimeoutException;
  29 import java.net.URL;
  30 import java.net.UnknownHostException;
  31 import java.util.ArrayList;
  32 import java.util.Collections;
  33 import java.util.HashMap;
  34 import java.util.List;
  35 import java.util.Map;
  36 import java.util.Random;
  37 import java.util.Set;
  38 import java.util.concurrent.CopyOnWriteArrayList;
  39 import java.util.concurrent.ExecutorService;
  40 import java.util.concurrent.Executors;
  41 import java.util.concurrent.Future;
  42 import java.util.concurrent.ScheduledFuture;
  43 import java.util.concurrent.TimeUnit;
  44 import java.util.regex.Pattern;
  45 import java.util.zip.GZIPInputStream;
  46
  47 import org.apache.commons.lang.StringUtils;
  48 import org.openhab.core.common.NamedThreadFactory;
  49 import org.openhab.core.thing.Bridge;
  50 import org.openhab.core.thing.ChannelUID;
  51 import org.openhab.core.thing.Thing;
  52 import org.openhab.core.thing.ThingStatus;
  53 import org.openhab.core.thing.ThingStatusDetail;
  54 import org.openhab.core.thing.ThingTypeUID;
  55 import org.openhab.core.thing.binding.BaseBridgeHandler;
  56 import org.openhab.core.types.Command;
  57 import org.openhab.core.types.RefreshType;
  58 import org.slf4j.Logger;
  59 import org.slf4j.LoggerFactory;
  60
  61 import com.google.gson.Gson;
  62 import com.google.gson.JsonArray;
  63 import com.google.gson.JsonElement;
  64 import com.google.gson.JsonObject;
  65 import com.google.gson.JsonParser;
  66
  67 /**
  68  * The {@link MieleBridgeHandler} is responsible for handling commands, which are
  69  * sent to one of the channels.
  70  *
  71  * @author Karel Goderis - Initial contribution
  72  * @author Kai Kreuzer - Fixed lifecycle issues
  73  * @author Martin Lepsy - Added protocol information to support WiFi devices & some refactoring for HomeDevice
  74  */
  75 public class MieleBridgeHandler extends BaseBridgeHandler {
  76
  77     public static final Set<ThingTypeUID> SUPPORTED_THING_TYPES = Collections.singleton(THING_TYPE_XGW3000);
  78
  79     private static final Pattern IP_PATTERN = Pattern
  80             .compile("^(([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.){3}([01]?\\d\\d?|2[0-4]\\d|25[0-5])$");
  81
  82     protected static final int POLLING_PERIOD = 15; // in seconds
  83     protected static final int JSON_RPC_PORT = 2810;
  84     protected static final String JSON_RPC_MULTICAST_IP1 = "239.255.68.139";
  85     protected static final String JSON_RPC_MULTICAST_IP2 = "224.255.68.139";
  86     private boolean lastBridgeConnectionState = false;
  87     private boolean currentBridgeConnectionState = false;
  88
  89     protected Random rand = new Random();
  90     protected Gson gson = new Gson();
  91     private final Logger logger = LoggerFactory.getLogger(MieleBridgeHandler.class);
  92
  93     protected List<ApplianceStatusListener> applianceStatusListeners = new CopyOnWriteArrayList<>();
  94     protected ScheduledFuture<?> pollingJob;
  95     protected ExecutorService executor;
  96     protected Future<?> eventListenerJob;
  97
  98     protected List<HomeDevice> previousHomeDevices = new CopyOnWriteArrayList<>();
  99
 100     protected URL url;
 101     protected Map<String, String> headers;
 102
 103     // Data structures to de-JSONify whatever Miele appliances are sending us
 104     public class HomeDevice {
 105
 106         private static final String PROTOCOL_LAN = "LAN";
 107
 108         public String Name;
 109         public String Status;
 110         public String ParentUID;
 111         public String ProtocolAdapterName;
 112         public String Vendor;
 113         public String UID;
 114         public String Type;
 115         public JsonArray DeviceClasses;
 116         public String Version;
 117         public String TimestampAdded;
 118         public JsonObject Error;
 119         public JsonObject Properties;
 120
 121         HomeDevice() {
 122         }
 123
 124         public String getId() {
 125             return getApplianceId().replaceAll("[^a-zA-Z0-9_]", "_");
 126         }
 127
 128         public String getProtocol() {
 129             return ProtocolAdapterName.equals(PROTOCOL_LAN) ? HDM_LAN : HDM_ZIGBEE;
 130         }
 131
 132         public String getApplianceId() {
 133             return ProtocolAdapterName.equals(PROTOCOL_LAN) ? StringUtils.right(UID, UID.length() - HDM_LAN.length())
 134                     : StringUtils.right(UID, UID.length() - HDM_ZIGBEE.length());
 135         }
 136     }
 137
 138     public class DeviceClassObject {
 139         public String DeviceClassType;
 140         public JsonArray Operations;
 141         public String DeviceClass;
 142         public JsonArray Properties;
 143
 144         DeviceClassObject() {
 145         }
 146     }
 147
 148     public class DeviceOperation {
 149         public String Name;
 150         public String Arguments;
 151         public JsonObject Metadata;
 152
 153         DeviceOperation() {
 154         }
 155     }
 156
 157     public class DeviceProperty {
 158         public String Name;
 159         public String Value;
 160         public int Polling;
 161         public JsonObject Metadata;
 162
 163         DeviceProperty() {
 164         }
 165     }
 166
 167     public class DeviceMetaData {
 168         public String Filter;
 169         public String description;
 170         public String LocalizedID;
 171         public String LocalizedValue;
 172         public JsonObject MieleEnum;
 173         public String access;
 174     }
 175
 176     public MieleBridgeHandler(Bridge bridge) {
 177         super(bridge);
 178     }
 179
 180     @Override
 181     public void initialize() {
 182         logger.debug("Initializing the Miele bridge handler.");
 183
 184         if (getConfig().get(HOST) != null && getConfig().get(INTERFACE) != null) {
 185             if (IP_PATTERN.matcher((String) getConfig().get(HOST)).matches()
 186                     && IP_PATTERN.matcher((String) getConfig().get(INTERFACE)).matches()) {
 187                 try {
 188                     url = new URL("http://" + (String) getConfig().get(HOST) + "/remote/json-rpc");
 189                 } catch (MalformedURLException e) {
 190                     logger.debug("An exception occurred while defining an URL :'{}'", e.getMessage());
 191                     updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.CONFIGURATION_ERROR, e.getMessage());
 192                     return;
 193                 }
 194
 195                 // for future usage - no headers to be set for now
 196                 headers = new HashMap<>();
 197
 198                 onUpdate();
 199                 updateStatus(ThingStatus.UNKNOWN);
 200             } else {
 201                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.CONFIGURATION_ERROR,
 202                         "Invalid IP address for the Miele@Home gateway or multicast interface:" + getConfig().get(HOST)
 203                                 + "/" + getConfig().get(INTERFACE));
 204             }
 205         } else {
 206             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.CONFIGURATION_ERROR,
 207                     "Cannot connect to the Miele gateway. host IP address or multicast interface are not set.");
 208         }
 209     }
 210
 211     private Runnable pollingRunnable = new Runnable() {
 212         @Override
 213         public void run() {
 214             if (IP_PATTERN.matcher((String) getConfig().get(HOST)).matches()) {
 215                 try {
 216                     if (isReachable((String) getConfig().get(HOST))) {
 217                         currentBridgeConnectionState = true;
 218                     } else {
 219                         currentBridgeConnectionState = false;
 220                         lastBridgeConnectionState = false;
 221                         onConnectionLost();
 222                     }
 223
 224                     if (!lastBridgeConnectionState && currentBridgeConnectionState) {
 225                         logger.debug("Connection to Miele Gateway {} established.", getConfig().get(HOST));
 226                         lastBridgeConnectionState = true;
 227                         onConnectionResumed();
 228                     }
 229
 230                     if (currentBridgeConnectionState) {
 231                         if (getThing().getStatus() == ThingStatus.ONLINE) {
 232                             List<HomeDevice> currentHomeDevices = getHomeDevices();
 233                             for (HomeDevice hd : currentHomeDevices) {
 234                                 boolean isExisting = false;
 235                                 for (HomeDevice phd : previousHomeDevices) {
 236                                     if (phd.UID.equals(hd.UID)) {
 237                                         isExisting = true;
 238                                         break;
 239                                     }
 240                                 }
 241                                 if (!isExisting) {
 242                                     logger.debug("A new appliance with ID '{}' has been added", hd.UID);
 243                                     for (ApplianceStatusListener listener : applianceStatusListeners) {
 244                                         listener.onApplianceAdded(hd);
 245                                     }
 246                                 }
 247                             }
 248
 249                             for (HomeDevice hd : previousHomeDevices) {
 250                                 boolean isCurrent = false;
 251                                 for (HomeDevice chd : currentHomeDevices) {
 252                                     if (chd.UID.equals(hd.UID)) {
 253                                         isCurrent = true;
 254                                         break;
 255                                     }
 256                                 }
 257                                 if (!isCurrent) {
 258                                     logger.debug("The appliance with ID '{}' has been removed", hd);
 259                                     for (ApplianceStatusListener listener : applianceStatusListeners) {
 260                                         listener.onApplianceRemoved(hd);
 261                                     }
 262                                 }
 263                             }
 264
 265                             previousHomeDevices = currentHomeDevices;
 266
 267                             for (Thing appliance : getThing().getThings()) {
 268                                 if (appliance.getStatus() == ThingStatus.ONLINE) {
 269                                     String UID = appliance.getProperties().get(PROTOCOL_PROPERTY_NAME)
 270                                             + (String) appliance.getConfiguration().getProperties().get(APPLIANCE_ID);
 271
 272                                     Object[] args = new Object[2];
 273                                     args[0] = UID;
 274                                     args[1] = true;
 275                                     JsonElement result = invokeRPC("HDAccess/getDeviceClassObjects", args);
 276
 277                                     if (result != null) {
 278                                         for (JsonElement obj : result.getAsJsonArray()) {
 279                                             try {
 280                                                 DeviceClassObject dco = gson.fromJson(obj, DeviceClassObject.class);
 281
 282                                                 for (ApplianceStatusListener listener : applianceStatusListeners) {
 283                                                     listener.onApplianceStateChanged(UID, dco);
 284                                                 }
 285                                             } catch (Exception e) {
 286                                                 logger.debug("An exception occurred while quering an appliance : '{}'",
 287                                                         e.getMessage());
 288                                             }
 289                                         }
 290                                     }
 291                                 }
 292                             }
 293                         }
 294                     }
 295                 } catch (Exception e) {
 296                     logger.debug("An exception occurred while polling an appliance :'{}'", e.getMessage());
 297                 }
 298             } else {
 299                 logger.debug("Invalid IP address for the Miele@Home gateway : '{}'", getConfig().get(HOST));
 300             }
 301         }
 302
 303         private boolean isReachable(String ipAddress) {
 304             try {
 305                 // note that InetAddress.isReachable is unreliable, see
 306                 // http://stackoverflow.com/questions/9922543/why-does-inetaddress-isreachable-return-false-when-i-can-ping-the-ip-address
 307                 // That's why we do an HTTP access instead
 308
 309                 // If there is no connection, this line will fail
 310                 JsonElement result = invokeRPC("system.listMethods", null);
 311                 if (result == null) {
 312                     logger.debug("{} is not reachable", ipAddress);
 313                     return false;
 314                 }
 315             } catch (Exception e) {
 316                 return false;
 317             }
 318
 319             logger.debug("{} is reachable", ipAddress);
 320             return true;
 321         }
 322     };
 323
 324     public List<HomeDevice> getHomeDevices() {
 325         List<HomeDevice> devices = new ArrayList<>();
 326
 327         if (getThing().getStatus() == ThingStatus.ONLINE) {
 328             try {
 329                 String[] args = new String[1];
 330                 args[0] = "(type=SuperVision)";
 331                 JsonElement result = invokeRPC("HDAccess/getHomeDevices", args);
 332
 333                 for (JsonElement obj : result.getAsJsonArray()) {
 334                     HomeDevice hd = gson.fromJson(obj, HomeDevice.class);
 335                     devices.add(hd);
 336                 }
 337             } catch (Exception e) {
 338                 logger.debug("An exception occurred while getting the home devices :'{}'", e.getMessage());
 339             }
 340         }
 341         return devices;
 342     }
 343
 344     private Runnable eventListenerRunnable = () -> {
 345         if (IP_PATTERN.matcher((String) getConfig().get(INTERFACE)).matches()) {
 346             while (true) {
 347                 // Get the address that we are going to connect to.
 348                 InetAddress address1 = null;
 349                 InetAddress address2 = null;
 350                 try {
 351                     address1 = InetAddress.getByName(JSON_RPC_MULTICAST_IP1);
 352                     address2 = InetAddress.getByName(JSON_RPC_MULTICAST_IP2);
 353                 } catch (UnknownHostException e) {
 354                     logger.debug("An exception occurred while setting up the multicast receiver : '{}'",
 355                             e.getMessage());
 356                 }
 357
 358                 byte[] buf = new byte[256];
 359                 MulticastSocket clientSocket = null;
 360
 361                 while (true) {
 362                     try {
 363                         clientSocket = new MulticastSocket(JSON_RPC_PORT);
 364                         clientSocket.setSoTimeout(100);
 365
 366                         clientSocket.setInterface(InetAddress.getByName((String) getConfig().get(INTERFACE)));
 367                         clientSocket.joinGroup(address1);
 368                         clientSocket.joinGroup(address2);
 369
 370                         while (true) {
 371                             try {
 372                                 buf = new byte[256];
 373                                 DatagramPacket packet = new DatagramPacket(buf, buf.length);
 374                                 clientSocket.receive(packet);
 375
 376                                 String event = new String(packet.getData());
 377                                 logger.debug("Received a multicast event '{}' from '{}:{}'", event, packet.getAddress(),
 378                                         packet.getPort());
 379
 380                                 DeviceProperty dp = new DeviceProperty();
 381                                 String uid = null;
 382
 383                                 String[] parts = StringUtils.split(event, "&");
 384                                 for (String p : parts) {
 385                                     String[] subparts = StringUtils.split(p, "=");
 386                                     switch (subparts[0]) {
 387                                         case "property": {
 388                                             dp.Name = subparts[1];
 389                                             break;
 390                                         }
 391                                         case "value": {
 392                                             dp.Value = subparts[1];
 393                                             break;
 394                                         }
 395                                         case "id": {
 396                                             uid = subparts[1];
 397                                             break;
 398                                         }
 399                                     }
 400                                 }
 401
 402                                 for (ApplianceStatusListener listener : applianceStatusListeners) {
 403                                     listener.onAppliancePropertyChanged(uid, dp);
 404                                 }
 405                             } catch (SocketTimeoutException e) {
 406                                 try {
 407                                     Thread.sleep(500);
 408                                 } catch (InterruptedException ex) {
 409                                     logger.debug("Eventlistener has been interrupted.");
 410                                     break;
 411                                 }
 412                             }
 413                         }
 414                     } catch (Exception ex) {
 415                         logger.debug("An exception occurred while receiving multicast packets : '{}'", ex.getMessage());
 416                     }
 417
 418                     // restart the cycle with a clean slate
 419                     try {
 420                         if (clientSocket != null) {
 421                             clientSocket.leaveGroup(address1);
 422                             clientSocket.leaveGroup(address2);
 423                         }
 424                     } catch (IOException e) {
 425                         logger.debug("An exception occurred while leaving multicast group : '{}'", e.getMessage());
 426                     }
 427                     if (clientSocket != null) {
 428                         clientSocket.close();
 429                     }
 430                 }
 431             }
 432         } else {
 433             logger.debug("Invalid IP address for the multicast interface : '{}'", getConfig().get(INTERFACE));
 434         }
 435     };
 436
 437     public JsonElement invokeOperation(String UID, String modelID, String methodName) {
 438         return invokeOperation(UID, modelID, methodName, HDM_ZIGBEE);
 439     }
 440
 441     public JsonElement invokeOperation(String UID, String modelID, String methodName, String protocol) {
 442         if (getThing().getStatus() == ThingStatus.ONLINE) {
 443             Object[] args = new Object[4];
 444             args[0] = protocol + UID;
 445             args[1] = "com.miele.xgw3000.gateway.hdm.deviceclasses.Miele" + modelID;
 446             args[2] = methodName;
 447             args[3] = null;
 448             return invokeRPC("HDAccess/invokeDCOOperation", args);
 449         } else {
 450             logger.debug("The Bridge is offline - operations can not be invoked.");
 451             return null;
 452         }
 453     }
 454
 455     protected JsonElement invokeRPC(String methodName, Object[] args) {
 456         int id = rand.nextInt(Integer.MAX_VALUE);
 457
 458         JsonObject req = new JsonObject();
 459         req.addProperty("jsonrpc", "2.0");
 460         req.addProperty("id", id);
 461         req.addProperty("method", methodName);
 462
 463         JsonElement result = null;
 464
 465         JsonArray params = new JsonArray();
 466         if (args != null) {
 467             for (Object o : args) {
 468                 params.add(gson.toJsonTree(o));
 469             }
 470         }
 471         req.add("params", params);
 472
 473         String requestData = req.toString();
 474         String responseData = null;
 475         try {
 476             responseData = post(url, headers, requestData);
 477         } catch (Exception e) {
 478             logger.debug("An exception occurred while posting data : '{}'", e.getMessage());
 479         }
 480
 481         if (responseData != null) {
 482             logger.debug("The request '{}' yields '{}'", requestData, responseData);
 483             JsonParser parser = new JsonParser();
 484             JsonObject resp = (JsonObject) parser.parse(new StringReader(responseData));
 485
 486             result = resp.get("result");
 487             JsonElement error = resp.get("error");
 488
 489             if (error != null && !error.isJsonNull()) {
 490                 if (error.isJsonPrimitive()) {
 491                     logger.debug("A remote exception occurred: '{}'", error.getAsString());
 492                 } else if (error.isJsonObject()) {
 493                     JsonObject o = error.getAsJsonObject();
 494                     Integer code = (o.has("code") ? o.get("code").getAsInt() : null);
 495                     String message = (o.has("message") ? o.get("message").getAsString() : null);
 496                     String data = (o.has("data") ? (o.get("data") instanceof JsonObject ? o.get("data").toString()
 497                             : o.get("data").getAsString()) : null);
 498                     logger.debug("A remote exception occurred: '{}':'{}':'{}'", code, message, data);
 499                 } else {
 500                     logger.debug("An unknown remote exception occurred: '{}'", error.toString());
 501                 }
 502             }
 503         }
 504
 505         return result;
 506     }
 507
 508     protected String post(URL url, Map<String, String> headers, String data) throws IOException {
 509         HttpURLConnection connection = (HttpURLConnection) url.openConnection();
 510
 511         if (headers != null) {
 512             for (Map.Entry<String, String> entry : headers.entrySet()) {
 513                 connection.addRequestProperty(entry.getKey(), entry.getValue());
 514             }
 515         }
 516
 517         connection.addRequestProperty("Accept-Encoding", "gzip");
 518
 519         connection.setRequestMethod("POST");
 520         connection.setDoOutput(true);
 521         connection.connect();
 522
 523         OutputStream out = null;
 524
 525         try {
 526             out = connection.getOutputStream();
 527
 528             out.write(data.getBytes());
 529             out.flush();
 530
 531             int statusCode = connection.getResponseCode();
 532             if (statusCode != HttpURLConnection.HTTP_OK) {
 533                 logger.debug("An unexpected status code was returned: '{}'", statusCode);
 534             }
 535         } finally {
 536             if (out != null) {
 537                 out.close();
 538             }
 539         }
 540
 541         String responseEncoding = connection.getHeaderField("Content-Encoding");
 542         responseEncoding = (responseEncoding == null ? "" : responseEncoding.trim());
 543
 544         ByteArrayOutputStream bos = new ByteArrayOutputStream();
 545
 546         InputStream in = connection.getInputStream();
 547         try {
 548             in = connection.getInputStream();
 549             if ("gzip".equalsIgnoreCase(responseEncoding)) {
 550                 in = new GZIPInputStream(in);
 551             }
 552             in = new BufferedInputStream(in);
 553
 554             byte[] buff = new byte[1024];
 555             int n;
 556             while ((n = in.read(buff)) > 0) {
 557                 bos.write(buff, 0, n);
 558             }
 559             bos.flush();
 560             bos.close();
 561         } finally {
 562             if (in != null) {
 563                 in.close();
 564             }
 565         }
 566
 567         return bos.toString();
 568     }
 569
 570     private synchronized void onUpdate() {
 571         logger.debug("Scheduling the Miele polling job");
 572         if (pollingJob == null || pollingJob.isCancelled()) {
 573             logger.trace("Scheduling the Miele polling job period is {}", POLLING_PERIOD);
 574             pollingJob = scheduler.scheduleWithFixedDelay(pollingRunnable, 0, POLLING_PERIOD, TimeUnit.SECONDS);
 575             logger.trace("Scheduling the Miele polling job Job is done ?{}", pollingJob.isDone());
 576         }
 577         logger.debug("Scheduling the Miele event listener job");
 578
 579         if (eventListenerJob == null || eventListenerJob.isCancelled()) {
 580             executor = Executors.newSingleThreadExecutor(new NamedThreadFactory("binding-miele"));
 581             eventListenerJob = executor.submit(eventListenerRunnable);
 582         }
 583     }
 584
 585     /**
 586      * This method is called whenever the connection to the given {@link MieleBridge} is lost.
 587      *
 588      */
 589     public void onConnectionLost() {
 590         updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.COMMUNICATION_ERROR);
 591     }
 592
 593     /**
 594      * This method is called whenever the connection to the given {@link MieleBridge} is resumed.
 595      *
 596      * @param bridge the hue bridge the connection is resumed to
 597      */
 598     public void onConnectionResumed() {
 599         updateStatus(ThingStatus.ONLINE);
 600         for (Thing thing : getThing().getThings()) {
 601             MieleApplianceHandler<?> handler = (MieleApplianceHandler<?>) thing.getHandler();
 602             if (handler != null) {
 603                 handler.onBridgeConnectionResumed();
 604             }
 605         }
 606     }
 607
 608     public boolean registerApplianceStatusListener(ApplianceStatusListener applianceStatusListener) {
 609         if (applianceStatusListener == null) {
 610             throw new IllegalArgumentException("It's not allowed to pass a null ApplianceStatusListener.");
 611         }
 612         boolean result = applianceStatusListeners.add(applianceStatusListener);
 613         if (result && isInitialized()) {
 614             onUpdate();
 615
 616             for (HomeDevice hd : getHomeDevices()) {
 617                 applianceStatusListener.onApplianceAdded(hd);
 618             }
 619         }
 620         return result;
 621     }
 622
 623     public boolean unregisterApplianceStatusListener(ApplianceStatusListener applianceStatusListener) {
 624         boolean result = applianceStatusListeners.remove(applianceStatusListener);
 625         if (result && isInitialized()) {
 626             onUpdate();
 627         }
 628         return result;
 629     }
 630
 631     @Override
 632     public void handleCommand(ChannelUID channelUID, Command command) {
 633         // Nothing to do here - the XGW bridge does not handle commands, for now
 634         if (command instanceof RefreshType) {
 635             // Placeholder for future refinement
 636             return;
 637         }
 638     }
 639
 640     @Override
 641     public void dispose() {
 642         super.dispose();
 643         if (pollingJob != null) {
 644             pollingJob.cancel(true);
 645             pollingJob = null;
 646         }
 647         if (eventListenerJob != null) {
 648             eventListenerJob.cancel(true);
 649             eventListenerJob = null;
 650         }
 651         if (executor != null) {
 652             executor.shutdownNow();
 653             executor = null;
 654         }
 655     }
 656 }