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