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

   1 /**
   2  * Copyright (c) 2010-2024 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.samsungtv.internal.service;
  14
  15 import static org.openhab.binding.samsungtv.internal.SamsungTvBindingConstants.*;
  16 import static org.openhab.binding.samsungtv.internal.config.SamsungTvConfiguration.*;
  17
  18 import java.io.ByteArrayInputStream;
  19 import java.io.IOException;
  20 import java.io.InputStream;
  21 import java.net.URI;
  22 import java.net.URISyntaxException;
  23 import java.net.http.HttpClient;
  24 import java.net.http.HttpRequest;
  25 import java.net.http.HttpResponse;
  26 import java.net.http.HttpResponse.BodyHandler;
  27 import java.net.http.HttpResponse.BodySubscriber;
  28 import java.net.http.HttpResponse.ResponseInfo;
  29 import java.nio.ByteBuffer;
  30 import java.nio.charset.StandardCharsets;
  31 import java.time.Duration;
  32 import java.util.ArrayList;
  33 import java.util.Arrays;
  34 import java.util.Collections;
  35 import java.util.HashMap;
  36 import java.util.List;
  37 import java.util.Map;
  38 import java.util.Optional;
  39 import java.util.Properties;
  40 import java.util.concurrent.CompletableFuture;
  41 import java.util.concurrent.CompletionStage;
  42 import java.util.concurrent.CountDownLatch;
  43 import java.util.concurrent.ExecutionException;
  44 import java.util.concurrent.Flow.Subscription;
  45 import java.util.concurrent.TimeUnit;
  46 import java.util.concurrent.TimeoutException;
  47 import java.util.stream.Collectors;
  48 import java.util.stream.IntStream;
  49
  50 import org.eclipse.jdt.annotation.NonNullByDefault;
  51 import org.eclipse.jdt.annotation.Nullable;
  52 import org.eclipse.jetty.http.HttpMethod;
  53 import org.openhab.binding.samsungtv.internal.handler.SamsungTvHandler;
  54 import org.openhab.binding.samsungtv.internal.service.api.SamsungTvService;
  55 import org.openhab.core.io.net.http.HttpUtil;
  56 import org.openhab.core.library.types.DecimalType;
  57 import org.openhab.core.library.types.StringType;
  58 import org.openhab.core.types.Command;
  59 import org.openhab.core.types.RefreshType;
  60 import org.slf4j.Logger;
  61 import org.slf4j.LoggerFactory;
  62
  63 import com.google.gson.Gson;
  64 import com.google.gson.JsonArray;
  65 import com.google.gson.JsonElement;
  66 import com.google.gson.JsonObject;
  67 import com.google.gson.JsonSyntaxException;
  68 import com.google.gson.annotations.SerializedName;
  69
  70 /**
  71  * The {@link SmartThingsApiService} is responsible for handling the Smartthings cloud interface
  72  *
  73  *
  74  * @author Nick Waterton - Initial contribution
  75  */
  76 @NonNullByDefault
  77 public class SmartThingsApiService implements SamsungTvService {
  78
  79     public static final String SERVICE_NAME = "SmartthingsApi";
  80     private static final List<String> SUPPORTED_CHANNELS = Arrays.asList(SOURCE_NAME, SOURCE_ID);
  81     private static final List<String> REFRESH_CHANNELS = Arrays.asList(CHANNEL, CHANNEL_NAME, SOURCE_NAME, SOURCE_ID);
  82     // Smarttings URL
  83     private static final String SMARTTHINGS_URL = "api.smartthings.com";
  84     // Path for the information endpoint note the final /
  85     private static final String API_ENDPOINT_V1 = "/v1/";
  86     // private static final String INPUT_SOURCE = "/components/main/capabilities/mediaInputSource/status";
  87     // private static final String CURRENT_CHANNEL = "/components/main/capabilities/tvChannel/status";
  88     private static final String COMPONENTS = "/components/main/status";
  89     private static final String DEVICES = "devices";
  90     private static final String COMMAND = "/commands";
  91
  92     private final Logger logger = LoggerFactory.getLogger(SmartThingsApiService.class);
  93
  94     private String host = "";
  95     private String apiKey = "";
  96     private String deviceId = "";
  97     private int RATE_LIMIT = 1000;
  98     private int TIMEOUT = 1000; // connection timeout in ms
  99     private long prevUpdate = 0;
 100     private boolean online = false;
 101     private int errorCount = 0;
 102     private int MAX_ERRORS = 100;
 103
 104     private final SamsungTvHandler handler;
 105
 106     private Optional<TvValues> tvInfo = Optional.empty();
 107     private boolean subscriptionRunning = false;
 108     private Optional<BodyHandlerWrapper> handlerWrapper = Optional.empty();
 109     private Optional<STSubscription> subscription = Optional.empty();
 110
 111     private Map<String, Object> stateMap = Collections.synchronizedMap(new HashMap<>());
 112
 113     public SmartThingsApiService(String host, SamsungTvHandler handler) {
 114         this.handler = handler;
 115         this.host = host;
 116         this.apiKey = handler.configuration.getSmartThingsApiKey();
 117         this.deviceId = handler.configuration.getSmartThingsDeviceId();
 118         logger.debug("{}: Creating a Samsung TV Smartthings Api service", host);
 119     }
 120
 121     @Override
 122     public String getServiceName() {
 123         return SERVICE_NAME;
 124     }
 125
 126     @Override
 127     public List<String> getSupportedChannelNames(boolean refresh) {
 128         if (refresh) {
 129             if (subscriptionRunning) {
 130                 return Arrays.asList();
 131             }
 132             return REFRESH_CHANNELS;
 133         }
 134         logger.trace("{}: getSupportedChannelNames: {}", host, SUPPORTED_CHANNELS);
 135         return SUPPORTED_CHANNELS;
 136     }
 137
 138     // Description of tvValues
 139     @NonNullByDefault({})
 140     class TvValues {
 141         class MediaInputSource {
 142             ValuesList supportedInputSources;
 143             ValuesListMap supportedInputSourcesMap;
 144             Values inputSource;
 145         }
 146
 147         class TvChannel {
 148             Values tvChannel;
 149             Values tvChannelName;
 150         }
 151
 152         class Values {
 153             String value;
 154             String timestamp;
 155         }
 156
 157         class ValuesList {
 158             String[] value;
 159             String timestamp;
 160         }
 161
 162         class ValuesListMap {
 163             InputList[] value;
 164             String timestamp;
 165
 166             public String[] getInputList() {
 167                 return Optional.ofNullable(value).map(a -> Arrays.stream(a).map(b -> b.getId()).toArray(String[]::new))
 168                         .orElse(new String[0]);
 169             }
 170         }
 171
 172         class InputList {
 173             public String id;
 174             String name;
 175
 176             public String getId() {
 177                 return Optional.ofNullable(id).orElse("");
 178             }
 179         }
 180
 181         class Items {
 182             String deviceId;
 183             String name;
 184             String label;
 185
 186             public String getDeviceId() {
 187                 return Optional.ofNullable(deviceId).orElse("");
 188             }
 189
 190             public String getName() {
 191                 return Optional.ofNullable(name).orElse("");
 192             }
 193
 194             public String getLabel() {
 195                 return Optional.ofNullable(label).orElse("");
 196             }
 197         }
 198
 199         class Error {
 200             String code;
 201             String message;
 202             Details[] details;
 203         }
 204
 205         class Details {
 206             String code;
 207             String target;
 208             String message;
 209         }
 210
 211         @SerializedName(value = "samsungvd.mediaInputSource", alternate = { "mediaInputSource" })
 212         MediaInputSource mediaInputSource;
 213         TvChannel tvChannel;
 214         Items[] items;
 215         Error error;
 216
 217         public void updateSupportedInputSources(String[] values) {
 218             mediaInputSource.supportedInputSources.value = values;
 219         }
 220
 221         public Items[] getItems() {
 222             return Optional.ofNullable(items).orElse(new Items[0]);
 223         }
 224
 225         public String[] getSources() {
 226             return Optional.ofNullable(mediaInputSource).map(a -> a.supportedInputSources).map(a -> a.value)
 227                     .orElseGet(() -> getSourcesFromMap());
 228         }
 229
 230         public String[] getSourcesFromMap() {
 231             return Optional.ofNullable(mediaInputSource).map(a -> a.supportedInputSourcesMap).map(a -> a.getInputList())
 232                     .orElse(new String[0]);
 233         }
 234
 235         public String getSourcesString() {
 236             return Arrays.asList(getSources()).stream().collect(Collectors.joining(","));
 237         }
 238
 239         public String getInputSource() {
 240             return Optional.ofNullable(mediaInputSource).map(a -> a.inputSource).map(a -> a.value).orElse("");
 241         }
 242
 243         public int getInputSourceId() {
 244             return IntStream.range(0, getSources().length).filter(i -> getSources()[i].equals(getInputSource()))
 245                     .findFirst().orElse(-1);
 246         }
 247
 248         public Number getTvChannel() {
 249             return Optional.ofNullable(tvChannel).map(a -> a.tvChannel).map(a -> a.value).filter(i -> !i.isBlank())
 250                     .map(j -> parseTVChannel(j)).orElse(-1f);
 251         }
 252
 253         public String getTvChannelName() {
 254             return Optional.ofNullable(tvChannel).map(a -> a.tvChannelName).map(a -> a.value).orElse("");
 255         }
 256
 257         public boolean isError() {
 258             return Optional.ofNullable(error).isPresent();
 259         }
 260
 261         public String getError() {
 262             String code = Optional.ofNullable(error).map(a -> a.code).orElse("");
 263             String message = Optional.ofNullable(error).map(a -> a.message).orElse("");
 264             return String.format("%s, %s", code, message);
 265         }
 266     }
 267
 268     @NonNullByDefault({})
 269     class JSONContent {
 270         public JSONContent(String capability, String action, String value) {
 271             Command command = new Command();
 272             command.capability = capability;
 273             command.command = action;
 274             command.arguments = new String[] { value };
 275             commands = new Command[] { command };
 276         }
 277
 278         class Command {
 279             String component = "main";
 280             String capability;
 281             String command;
 282             String[] arguments;
 283         }
 284
 285         Command[] commands;
 286     }
 287
 288     @NonNullByDefault({})
 289     class JSONSubscriptionFilter {
 290         public JSONSubscriptionFilter(String deviceId) {
 291             SubscriptionFilter sub = new SubscriptionFilter();
 292             sub.value = new String[] { deviceId };
 293             subscriptionFilters = new SubscriptionFilter[] { sub };
 294         }
 295
 296         class SubscriptionFilter {
 297             String type = "DEVICEIDS";
 298             String[] value;
 299         }
 300
 301         SubscriptionFilter[] subscriptionFilters;
 302         String name = "OpenHAB Subscription";
 303     }
 304
 305     @NonNullByDefault({})
 306     class STSubscription {
 307
 308         String subscriptionId;
 309         String registrationUrl;
 310         String name;
 311         Integer version;
 312         SubscriptionFilters[] subscriptionFilters;
 313
 314         class SubscriptionFilters {
 315             String type;
 316             String[] value;
 317         }
 318
 319         public String getSubscriptionId() {
 320             return Optional.ofNullable(subscriptionId).orElse("");
 321         }
 322
 323         public String getregistrationUrl() {
 324             return Optional.ofNullable(registrationUrl).orElse("");
 325         }
 326     }
 327
 328     @NonNullByDefault({})
 329     class STSSEData {
 330
 331         long eventTime;
 332         String eventType;
 333         DeviceEvent deviceEvent;
 334         Optional<TvValues> tvInfo = Optional.empty();
 335
 336         class DeviceEvent {
 337
 338             String eventId;
 339             String locationId;
 340             String ownerId;
 341             String ownerType;
 342             String deviceId;
 343             String componentId;
 344             String capability; // example "sec.diagnosticsInformation"
 345             String attribute; // example "dumpType"
 346             JsonElement value; // example "id" or can be an array
 347             String valueType;
 348             boolean stateChange;
 349             JsonElement data;
 350             String subscriptionName;
 351
 352             class ValuesList {
 353                 // Array of supportedInputSourcesMap
 354                 String id;
 355                 String name;
 356
 357                 public String getId() {
 358                     return Optional.ofNullable(id).orElse("");
 359                 }
 360
 361                 public String getName() {
 362                     return Optional.ofNullable(name).orElse("");
 363                 }
 364
 365                 @Override
 366                 public String toString() {
 367                     return Map.of("id", getId(), "name", getName()).toString();
 368                 }
 369             }
 370
 371             public String getCapability() {
 372                 return Optional.ofNullable(capability).orElse("");
 373             }
 374
 375             public String getAttribute() {
 376                 return Optional.ofNullable(attribute).orElse("");
 377             }
 378
 379             public String getValueType() {
 380                 return Optional.ofNullable(valueType).orElse("");
 381             }
 382
 383             public List<?> getValuesAsList() throws JsonSyntaxException {
 384                 if ("array".equals(getValueType())) {
 385                     JsonArray resultArray = Optional.ofNullable((JsonArray) value.getAsJsonArray())
 386                             .orElse(new JsonArray());
 387                     try {
 388                         if (resultArray.get(0) instanceof JsonObject) {
 389                             // Only for Array of supportedInputSourcesMap
 390                             ValuesList[] values = new Gson().fromJson(resultArray, ValuesList[].class);
 391                             List<ValuesList> result = Optional.ofNullable(values).map(a -> Arrays.asList(a))
 392                                     .orElse(new ArrayList<ValuesList>());
 393                             return Optional.ofNullable(result).orElse(List.of());
 394                         } else {
 395                             List<String> result = new Gson().fromJson(resultArray, ArrayList.class);
 396                             return Optional.ofNullable(result).orElse(List.of());
 397                         }
 398                     } catch (IllegalStateException e) {
 399                     }
 400                 }
 401                 return List.of();
 402             }
 403
 404             public String getValue() {
 405                 if ("string".equals(getValueType())) {
 406                     return Optional.ofNullable((String) value.getAsString()).orElse("");
 407                 }
 408                 return "";
 409             }
 410         }
 411
 412         public void setTvInfo(Optional<TvValues> tvInfo) {
 413             this.tvInfo = tvInfo;
 414         }
 415
 416         public boolean getCapabilityAttribute(String capability, String attribute) {
 417             return Optional.ofNullable(deviceEvent).map(a -> a.getCapability()).filter(a -> a.equals(capability))
 418                     .isPresent()
 419                     && Optional.ofNullable(deviceEvent).map(a -> a.getAttribute()).filter(a -> a.equals(attribute))
 420                             .isPresent();
 421         }
 422
 423         public String getSwitch() {
 424             if (getCapabilityAttribute("switch", "switch")) {
 425                 return Optional.ofNullable(deviceEvent).map(a -> a.getValue()).orElse("");
 426             }
 427             return "";
 428         }
 429
 430         public String getInputSource() {
 431             if (getCapabilityAttribute("mediaInputSource", "inputSource")
 432                     || getCapabilityAttribute("samsungvd.mediaInputSource", "inputSource")) {
 433                 return Optional.ofNullable(deviceEvent).map(a -> a.getValue()).orElse("");
 434             }
 435             return "";
 436         }
 437
 438         public String[] getInputSourceList() {
 439             if (getCapabilityAttribute("mediaInputSource", "supportedInputSources")) {
 440                 return deviceEvent.getValuesAsList().toArray(String[]::new);
 441             }
 442             return new String[0];
 443         }
 444
 445         public List<?> getInputSourceMapList() {
 446             if (getCapabilityAttribute("samsungvd.mediaInputSource", "supportedInputSourcesMap")) {
 447                 return deviceEvent.getValuesAsList();
 448             }
 449             return List.of();
 450         }
 451
 452         public int getInputSourceId() {
 453             return this.tvInfo.map(t -> IntStream.range(0, t.getSources().length)
 454                     .filter(i -> t.getSources()[i].equals(getInputSource())).findFirst().orElse(-1)).orElse(-1);
 455         }
 456
 457         public Number getTvChannel() {
 458             if (getCapabilityAttribute("tvChannel", "tvChannel")) {
 459                 return Optional.ofNullable(deviceEvent).map(a -> a.getValue()).filter(i -> !i.isBlank())
 460                         .map(j -> parseTVChannel(j)).orElse(-1f);
 461             }
 462             return -1;
 463         }
 464
 465         public String getTvChannelName() {
 466             if (getCapabilityAttribute("tvChannel", "tvChannelName")) {
 467                 return Optional.ofNullable(deviceEvent).map(a -> a.getValue()).orElse("");
 468             }
 469             return "";
 470         }
 471     }
 472
 473     public Number parseTVChannel(String channel) {
 474         try {
 475             return Optional.ofNullable(channel)
 476                     .map(a -> a.replaceAll("\\D+", ".").replaceFirst("^\\D*((\\d+\\.\\d+)|(\\d+)).*", "$1"))
 477                     .map(Float::parseFloat).orElse(-1f);
 478         } catch (NumberFormatException ignore) {
 479         }
 480         return -1;
 481     }
 482
 483     public void updateTV() {
 484         if (!tvInfo.isPresent()) {
 485             fetchdata();
 486             tvInfo.ifPresent(t -> {
 487                 updateState(CHANNEL_NAME, t.getTvChannelName());
 488                 updateState(CHANNEL, t.getTvChannel());
 489                 updateState(SOURCE_NAME, t.getInputSource());
 490                 updateState(SOURCE_ID, t.getInputSourceId());
 491             });
 492         }
 493     }
 494
 495     /**
 496      * Smartthings API HTTP interface
 497      * Currently rate limited to 350 requests/minute
 498      *
 499      * @param method the method "GET" or "POST"
 500      * @param uri as a URI
 501      * @param content to POST (or null)
 502      * @return response
 503      */
 504     public Optional<String> sendUrl(HttpMethod method, URI uri, @Nullable InputStream content) throws IOException {
 505         // need to add header "Authorization":"Bearer " + apiKey;
 506         Properties headers = new Properties();
 507         headers.put("Authorization", "Bearer " + this.apiKey);
 508         logger.trace("{}: Sending {}", host, uri.toURL().toString());
 509         Optional<String> response = Optional.ofNullable(HttpUtil.executeUrl(method.toString(), uri.toURL().toString(),
 510                 headers, content, "application/json", TIMEOUT));
 511         if (!response.isPresent()) {
 512             throw new IOException("No Data");
 513         }
 514         response.ifPresent(r -> logger.trace("{}: Got response: {}", host, r));
 515         response.filter(r -> !r.startsWith("{")).ifPresent(r -> logger.debug("{}: Got response: {}", host, r));
 516         return response;
 517     }
 518
 519     /**
 520      * Smartthings API HTTP getter
 521      * Currently rate limited to 350 requests/minute
 522      *
 523      * @param value the query to send
 524      * @return tvValues
 525      */
 526     public synchronized Optional<TvValues> fetchTVProperties(String value) {
 527         if (apiKey.isBlank()) {
 528             return Optional.empty();
 529         }
 530         Optional<TvValues> tvValues = Optional.empty();
 531         try {
 532             String api = API_ENDPOINT_V1 + ((deviceId.isBlank()) ? "" : "devices/") + deviceId + value;
 533             URI uri = new URI("https", null, SMARTTHINGS_URL, 443, api, null, null);
 534             Optional<String> response = sendUrl(HttpMethod.GET, uri, null);
 535             tvValues = response.map(r -> new Gson().fromJson(r, TvValues.class));
 536             if (!tvValues.isPresent()) {
 537                 throw new IOException("No Data - is DeviceID correct?");
 538             }
 539             tvValues.filter(t -> t.isError()).ifPresent(t -> logger.debug("{}: Error: {}", host, t.getError()));
 540             errorCount = 0;
 541         } catch (JsonSyntaxException | URISyntaxException | IOException e) {
 542             logger.debug("{}: Cannot connect to Smartthings Cloud: {}", host, e.getMessage());
 543             if (errorCount++ > MAX_ERRORS) {
 544                 logger.warn("{}: Too many connection errors, disabling SmartThings", host);
 545                 stop();
 546             }
 547         }
 548         return tvValues;
 549     }
 550
 551     /**
 552      * Smartthings API HTTP setter
 553      * Currently rate limited to 350 requests/minute
 554      *
 555      * @param capability eg mediaInputSource
 556      * @param command eg setInputSource
 557      * @param value from acceptible list eg HDMI1, digitalTv, AM etc
 558      * @return boolean true if successful
 559      */
 560     public synchronized boolean setTVProperties(String capability, String command, String value) {
 561         if (apiKey.isBlank() || deviceId.isBlank()) {
 562             return false;
 563         }
 564         Optional<String> response = Optional.empty();
 565         try {
 566             String contentString = new Gson().toJson(new JSONContent(capability, command, value));
 567             logger.trace("{}: content: {}", host, contentString);
 568             InputStream content = new ByteArrayInputStream(contentString.getBytes());
 569             String api = API_ENDPOINT_V1 + "devices/" + deviceId + COMMAND;
 570             URI uri = new URI("https", null, SMARTTHINGS_URL, 443, api, null, null);
 571             response = sendUrl(HttpMethod.POST, uri, content);
 572         } catch (JsonSyntaxException | URISyntaxException | IOException e) {
 573             logger.debug("{}: Send Command to Smartthings Cloud failed: {}", host, e.getMessage());
 574         }
 575         return response.map(r -> r.contains("ACCEPTED") || r.contains("COMPLETED")).orElse(false);
 576     }
 577
 578     /**
 579      * Smartthings API Subscription
 580      * Retrieves the Smartthings API Subscription from a remote service, performing an API call
 581      *
 582      * @return stSub
 583      */
 584     public synchronized Optional<STSubscription> smartthingsSubscription() {
 585         if (apiKey.isBlank() || deviceId.isBlank()) {
 586             return Optional.empty();
 587         }
 588         Optional<STSubscription> stSub = Optional.empty();
 589         try {
 590             logger.info("{}: SSE Creating Smartthings Subscription", host);
 591             String contentString = new Gson().toJson(new JSONSubscriptionFilter(deviceId));
 592             logger.trace("{}: subscription: {}", host, contentString);
 593             InputStream subscriptionFilter = new ByteArrayInputStream(contentString.getBytes());
 594             URI uri = new URI("https", null, SMARTTHINGS_URL, 443, "/subscriptions", null, null);
 595             Optional<String> response = sendUrl(HttpMethod.POST, uri, subscriptionFilter);
 596             stSub = response.map(r -> new Gson().fromJson(r, STSubscription.class));
 597             if (!stSub.isPresent()) {
 598                 throw new IOException("No Data - is DeviceID correct?");
 599             }
 600         } catch (JsonSyntaxException | URISyntaxException | IOException e) {
 601             logger.warn("{}: SSE Subscription to Smartthings Cloud failed: {}", host, e.getMessage());
 602         }
 603         return stSub;
 604     }
 605
 606     public synchronized void startSSE() {
 607         if (!subscriptionRunning) {
 608             logger.trace("{}: SSE Starting job", host);
 609             subscription = smartthingsSubscription();
 610             logger.trace("{}: SSE got subscription ID: {}", host,
 611                     subscription.map(a -> a.getSubscriptionId()).orElse("None"));
 612             if (!subscription.map(a -> a.getSubscriptionId()).orElse("").isBlank()) {
 613                 receiveSSEEvents();
 614             }
 615         }
 616     }
 617
 618     public void stopSSE() {
 619         handlerWrapper.ifPresent(a -> {
 620             a.cancel();
 621             logger.trace("{}: SSE Stopping job", host);
 622             handlerWrapper = Optional.empty();
 623             subscriptionRunning = false;
 624         });
 625     }
 626
 627     /**
 628      * SubscriberWrapper needed to make async SSE stream cancelable
 629      *
 630      */
 631     @NonNullByDefault({})
 632     private static class SubscriberWrapper implements BodySubscriber<Void> {
 633         private final CountDownLatch latch;
 634         private final BodySubscriber<Void> subscriber;
 635         private Subscription subscription;
 636
 637         private SubscriberWrapper(BodySubscriber<Void> subscriber, CountDownLatch latch) {
 638             this.subscriber = subscriber;
 639             this.latch = latch;
 640         }
 641
 642         @Override
 643         public CompletionStage<Void> getBody() {
 644             return subscriber.getBody();
 645         }
 646
 647         @Override
 648         public void onSubscribe(Subscription subscription) {
 649             subscriber.onSubscribe(subscription);
 650             this.subscription = subscription;
 651             latch.countDown();
 652         }
 653
 654         @Override
 655         public void onNext(List<ByteBuffer> item) {
 656             subscriber.onNext(item);
 657         }
 658
 659         @Override
 660         public void onError(Throwable throwable) {
 661             subscriber.onError(throwable);
 662         }
 663
 664         @Override
 665         public void onComplete() {
 666             subscriber.onComplete();
 667         }
 668
 669         public void cancel() {
 670             subscription.cancel();
 671         }
 672     }
 673
 674     @NonNullByDefault({})
 675     private static class BodyHandlerWrapper implements BodyHandler<Void> {
 676         private final CountDownLatch latch = new CountDownLatch(1);
 677         private final BodyHandler<Void> handler;
 678         private SubscriberWrapper subscriberWrapper;
 679         private int statusCode = -1;
 680
 681         private BodyHandlerWrapper(BodyHandler<Void> handler) {
 682             this.handler = handler;
 683         }
 684
 685         @Override
 686         public BodySubscriber<Void> apply(ResponseInfo responseInfo) {
 687             subscriberWrapper = new SubscriberWrapper(handler.apply(responseInfo), latch);
 688             this.statusCode = responseInfo.statusCode();
 689             return subscriberWrapper;
 690         }
 691
 692         public void waitForEvent(boolean cancel) {
 693             try {
 694                 CompletableFuture.runAsync(() -> {
 695                     try {
 696                         latch.await();
 697                         if (cancel) {
 698                             subscriberWrapper.cancel();
 699                         }
 700                     } catch (InterruptedException ignore) {
 701                     }
 702                 }).get(2, TimeUnit.SECONDS);
 703             } catch (InterruptedException | ExecutionException | TimeoutException ignore) {
 704             }
 705         }
 706
 707         public int getStatusCode() {
 708             waitForEvent(false);
 709             return statusCode;
 710         }
 711
 712         public void cancel() {
 713             waitForEvent(true);
 714         }
 715     }
 716
 717     public void receiveSSEEvents() {
 718         subscription.ifPresent(sub -> {
 719             updateTV();
 720             try {
 721                 URI uri = new URI(sub.getregistrationUrl());
 722                 HttpClient client = HttpClient.newHttpClient();
 723                 HttpRequest request = HttpRequest.newBuilder(uri).timeout(Duration.ofSeconds(2)).GET()
 724                         .header("Authorization", "Bearer " + this.apiKey).build();
 725                 handlerWrapper = Optional.ofNullable(
 726                         new BodyHandlerWrapper(HttpResponse.BodyHandlers.ofByteArrayConsumer(b -> processSSEEvent(b))));
 727                 handlerWrapper.ifPresent(h -> {
 728                     client.sendAsync(request, h);
 729                 });
 730                 logger.debug("{}: SSE job {}", host, checkResponseCode() ? "Started" : "Failed");
 731             } catch (URISyntaxException e) {
 732                 logger.warn("{}: SSE URI Exception: {}", host, e.getMessage());
 733             }
 734         });
 735     }
 736
 737     boolean checkResponseCode() {
 738         int respCode = handlerWrapper.map(a -> a.getStatusCode()).orElse(-1);
 739         logger.trace("{}: SSE GOT Response Code: {}", host, respCode);
 740         subscriptionRunning = (respCode == 200);
 741         return subscriptionRunning;
 742     }
 743
 744     Map<String, String> bytesToMap(byte[] bytes) {
 745         String s = new String(bytes, StandardCharsets.UTF_8);
 746         // logger.trace("{}: SSE received: {}", host, s);
 747         Map<String, String> properties = new HashMap<String, String>();
 748         String[] pairs = s.split("\r?\n");
 749         for (String pair : pairs) {
 750             String[] kv = pair.split(":", 2);
 751             properties.put(kv[0].trim(), kv[1].trim());
 752         }
 753         logger.trace("{}: SSE received: {}", host, properties);
 754         updateTV();
 755         return properties;
 756     }
 757
 758     synchronized void processSSEEvent(Optional<byte[]> bytes) {
 759         bytes.ifPresent(b -> {
 760             Map<String, String> properties = bytesToMap(b);
 761             String rawData = properties.getOrDefault("data", "none");
 762             String event = properties.getOrDefault("event", "none");
 763             // logger.trace("{}: SSE Decoding event: {}", host, event);
 764             switch (event) {
 765                 case "CONTROL_EVENT":
 766                     subscriptionRunning = "welcome".equals(rawData);
 767                     if (!subscriptionRunning) {
 768                         logger.trace("{}: SSE Subscription ended", host);
 769                         startSSE();
 770                     }
 771                     break;
 772                 case "DEVICE_EVENT":
 773                     try {
 774                         // decode json here
 775                         Optional<STSSEData> data = Optional.ofNullable(new Gson().fromJson(rawData, STSSEData.class));
 776                         data.ifPresentOrElse(d -> {
 777                             d.setTvInfo(tvInfo);
 778                             String[] inputList = d.getInputSourceList();
 779                             if (inputList.length > 0) {
 780                                 logger.trace("{}: SSE Got input source list: {}", host, Arrays.asList(inputList));
 781                                 tvInfo.ifPresent(a -> a.updateSupportedInputSources(inputList));
 782                             }
 783                             String inputSource = d.getInputSource();
 784                             if (!inputSource.isBlank()) {
 785                                 updateState(SOURCE_NAME, inputSource);
 786                                 int sourceId = d.getInputSourceId();
 787                                 logger.trace("{}: SSE Got input source: {} ID: {}", host, inputSource, sourceId);
 788                                 updateState(SOURCE_ID, sourceId);
 789                             }
 790                             Number tvChannel = d.getTvChannel();
 791                             if (tvChannel.intValue() != -1) {
 792                                 updateState(CHANNEL, tvChannel);
 793                                 String tvChannelName = d.getTvChannelName();
 794                                 logger.trace("{}: SSE Got TV Channel Name: {} Channel: {}", host, tvChannelName,
 795                                         tvChannel);
 796                                 updateState(CHANNEL_NAME, tvChannelName);
 797                             }
 798                             String Power = d.getSwitch();
 799                             if (!Power.isBlank()) {
 800                                 logger.debug("{}: SSE Got TV Power: {}", host, Power);
 801                                 if ("on".equals(Power)) {
 802                                     // handler.putOnline(); // ignore on event for now
 803                                 } else {
 804                                     // handler.setOffline(); // ignore off event for now
 805                                 }
 806                             }
 807                         }, () -> logger.warn("{}: SSE Received NULL data", host));
 808                     } catch (JsonSyntaxException e) {
 809                         logger.warn("{}: SmartThingsApiService: Error ({}) in message: {}", host, e.getMessage(),
 810                                 rawData);
 811                     }
 812                     break;
 813                 default:
 814                     logger.trace("{}: SSE not handling event: {}", host, event);
 815                     break;
 816             }
 817         });
 818     }
 819
 820     private boolean updateDeviceID(TvValues.Items item) {
 821         this.deviceId = item.getDeviceId();
 822         logger.debug("{}: found {} device, adding device id {}", host, item.getName(), deviceId);
 823         handler.putConfig(SMARTTHINGS_DEVICEID, deviceId);
 824         prevUpdate = 0;
 825         return true;
 826     }
 827
 828     public boolean fetchdata() {
 829         if (System.currentTimeMillis() >= prevUpdate + RATE_LIMIT) {
 830             if (deviceId.isBlank()) {
 831                 tvInfo = fetchTVProperties(DEVICES);
 832                 boolean found = false;
 833                 if (tvInfo.isPresent()) {
 834                     TvValues t = tvInfo.get();
 835                     switch (t.getItems().length) {
 836                         case 0:
 837                         case 1:
 838                             logger.warn("{}: No devices found - please add your TV to the Smartthings app", host);
 839                             break;
 840                         case 2:
 841                             found = Arrays.asList(t.getItems()).stream().filter(a -> "Samsung TV".equals(a.getName()))
 842                                     .map(a -> updateDeviceID(a)).findFirst().orElse(false);
 843                             break;
 844                         default:
 845                             logger.warn("{}: No device Id selected, please enter one of the following:", host);
 846                             Arrays.asList(t.getItems()).stream().forEach(a -> logger.info("{}: '{}' : {}({})", host,
 847                                     a.getDeviceId(), a.getName(), a.getLabel()));
 848                     }
 849                 }
 850                 if (found) {
 851                     return fetchdata();
 852                 } else {
 853                     stop();
 854                     return false;
 855                 }
 856             }
 857             tvInfo = fetchTVProperties(COMPONENTS);
 858             prevUpdate = System.currentTimeMillis();
 859         }
 860         return (tvInfo.isPresent());
 861     }
 862
 863     @Override
 864     public void start() {
 865         online = true;
 866         errorCount = 0;
 867         startSSE();
 868     }
 869
 870     @Override
 871     public void stop() {
 872         online = false;
 873         stopSSE();
 874     }
 875
 876     @Override
 877     public void clearCache() {
 878         stateMap.clear();
 879         start();
 880     }
 881
 882     @Override
 883     public boolean isUpnp() {
 884         return false;
 885     }
 886
 887     @Override
 888     public boolean checkConnection() {
 889         return online;
 890     }
 891
 892     @Override
 893     public boolean handleCommand(String channel, Command command) {
 894         logger.trace("{}: Received channel: {}, command: {}", host, channel, command);
 895         if (!checkConnection()) {
 896             logger.trace("{}: Smartthings offline", host);
 897             return false;
 898         }
 899
 900         if (fetchdata()) {
 901             return tvInfo.map(t -> {
 902                 boolean result = false;
 903                 if (command == RefreshType.REFRESH) {
 904                     switch (channel) {
 905                         case CHANNEL_NAME:
 906                             updateState(CHANNEL_NAME, t.getTvChannelName());
 907                             break;
 908                         case CHANNEL:
 909                             updateState(CHANNEL, t.getTvChannel());
 910                             break;
 911                         case SOURCE_ID:
 912                         case SOURCE_NAME:
 913                             updateState(SOURCE_NAME, t.getInputSource());
 914                             updateState(SOURCE_ID, t.getInputSourceId());
 915                             break;
 916                         default:
 917                             break;
 918                     }
 919                     return true;
 920                 }
 921
 922                 switch (channel) {
 923                     case SOURCE_ID:
 924                         if (command instanceof DecimalType commandAsDecimalType) {
 925                             int val = commandAsDecimalType.intValue();
 926                             if (val >= 0 && val < t.getSources().length) {
 927                                 result = setSourceName(t.getSources()[val]);
 928                             } else {
 929                                 logger.warn("{}: Invalid source ID: {}, acceptable: 0..{}", host, command,
 930                                         t.getSources().length);
 931                             }
 932                         }
 933                         break;
 934                     case SOURCE_NAME:
 935                         if (command instanceof StringType) {
 936                             if (t.getSourcesString().contains(command.toString()) || t.getSourcesString().isBlank()) {
 937                                 result = setSourceName(command.toString());
 938                             } else {
 939                                 logger.warn("{}: Invalid source Name: {}, acceptable: {}", host, command,
 940                                         t.getSourcesString());
 941                             }
 942                         }
 943                         break;
 944                     default:
 945                         logger.warn("{}: Samsung TV doesn't support transmitting for channel '{}'", host, channel);
 946                 }
 947                 if (!result) {
 948                     logger.warn("{}: Smartthings: wrong command type {} channel {}", host, command, channel);
 949                 }
 950                 return result;
 951             }).orElse(false);
 952         }
 953         return false;
 954     }
 955
 956     private void updateState(String channel, Object value) {
 957         if (!stateMap.getOrDefault(channel, "None").equals(value)) {
 958             switch (channel) {
 959                 case CHANNEL:
 960                 case SOURCE_ID:
 961                     handler.valueReceived(channel, new DecimalType((Number) value));
 962                     break;
 963                 default:
 964                     handler.valueReceived(channel, new StringType((String) value));
 965                     break;
 966             }
 967             stateMap.put(channel, value);
 968         } else {
 969             logger.trace("{}: Value '{}' for {} hasn't changed, ignoring update", host, value, channel);
 970         }
 971     }
 972
 973     private boolean setSourceName(String value) {
 974         return setTVProperties("mediaInputSource", "setInputSource", value);
 975     }
 976 }