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(@Nullable String channel) {
 474         try {
 475             return channel != null
 476                     ? Float.parseFloat(
 477                             channel.replaceAll("\\D+", ".").replaceFirst("^\\D*((\\d+\\.\\d+)|(\\d+)).*", "$1"))
 478                     : -1f;
 479         } catch (NumberFormatException ignore) {
 480         }
 481         return -1;
 482     }
 483
 484     public void updateTV() {
 485         if (!tvInfo.isPresent()) {
 486             fetchdata();
 487             tvInfo.ifPresent(t -> {
 488                 updateState(CHANNEL_NAME, t.getTvChannelName());
 489                 updateState(CHANNEL, t.getTvChannel());
 490                 updateState(SOURCE_NAME, t.getInputSource());
 491                 updateState(SOURCE_ID, t.getInputSourceId());
 492             });
 493         }
 494     }
 495
 496     /**
 497      * Smartthings API HTTP interface
 498      * Currently rate limited to 350 requests/minute
 499      *
 500      * @param method the method "GET" or "POST"
 501      * @param uri as a URI
 502      * @param content to POST (or null)
 503      * @return response
 504      */
 505     public Optional<String> sendUrl(HttpMethod method, URI uri, @Nullable InputStream content) throws IOException {
 506         // need to add header "Authorization":"Bearer " + apiKey;
 507         Properties headers = new Properties();
 508         headers.put("Authorization", "Bearer " + this.apiKey);
 509         logger.trace("{}: Sending {}", host, uri.toURL().toString());
 510         Optional<String> response = Optional.ofNullable(HttpUtil.executeUrl(method.toString(), uri.toURL().toString(),
 511                 headers, content, "application/json", TIMEOUT));
 512         if (!response.isPresent()) {
 513             throw new IOException("No Data");
 514         }
 515         response.ifPresent(r -> logger.trace("{}: Got response: {}", host, r));
 516         response.filter(r -> !r.startsWith("{")).ifPresent(r -> logger.debug("{}: Got response: {}", host, r));
 517         return response;
 518     }
 519
 520     /**
 521      * Smartthings API HTTP getter
 522      * Currently rate limited to 350 requests/minute
 523      *
 524      * @param value the query to send
 525      * @return tvValues
 526      */
 527     public synchronized Optional<TvValues> fetchTVProperties(String value) {
 528         if (apiKey.isBlank()) {
 529             return Optional.empty();
 530         }
 531         Optional<TvValues> tvValues = Optional.empty();
 532         try {
 533             String api = API_ENDPOINT_V1 + ((deviceId.isBlank()) ? "" : "devices/") + deviceId + value;
 534             URI uri = new URI("https", null, SMARTTHINGS_URL, 443, api, null, null);
 535             Optional<String> response = sendUrl(HttpMethod.GET, uri, null);
 536             tvValues = response.map(r -> new Gson().fromJson(r, TvValues.class));
 537             if (!tvValues.isPresent()) {
 538                 throw new IOException("No Data - is DeviceID correct?");
 539             }
 540             tvValues.filter(t -> t.isError()).ifPresent(t -> logger.debug("{}: Error: {}", host, t.getError()));
 541             errorCount = 0;
 542         } catch (JsonSyntaxException | URISyntaxException | IOException e) {
 543             logger.debug("{}: Cannot connect to Smartthings Cloud: {}", host, e.getMessage());
 544             if (errorCount++ > MAX_ERRORS) {
 545                 logger.warn("{}: Too many connection errors, disabling SmartThings", host);
 546                 stop();
 547             }
 548         }
 549         return tvValues;
 550     }
 551
 552     /**
 553      * Smartthings API HTTP setter
 554      * Currently rate limited to 350 requests/minute
 555      *
 556      * @param capability eg mediaInputSource
 557      * @param command eg setInputSource
 558      * @param value from acceptible list eg HDMI1, digitalTv, AM etc
 559      * @return boolean true if successful
 560      */
 561     public synchronized boolean setTVProperties(String capability, String command, String value) {
 562         if (apiKey.isBlank() || deviceId.isBlank()) {
 563             return false;
 564         }
 565         Optional<String> response = Optional.empty();
 566         try {
 567             String contentString = new Gson().toJson(new JSONContent(capability, command, value));
 568             logger.trace("{}: content: {}", host, contentString);
 569             InputStream content = new ByteArrayInputStream(contentString.getBytes());
 570             String api = API_ENDPOINT_V1 + "devices/" + deviceId + COMMAND;
 571             URI uri = new URI("https", null, SMARTTHINGS_URL, 443, api, null, null);
 572             response = sendUrl(HttpMethod.POST, uri, content);
 573         } catch (JsonSyntaxException | URISyntaxException | IOException e) {
 574             logger.debug("{}: Send Command to Smartthings Cloud failed: {}", host, e.getMessage());
 575         }
 576         return response.map(r -> r.contains("ACCEPTED") || r.contains("COMPLETED")).orElse(false);
 577     }
 578
 579     /**
 580      * Smartthings API Subscription
 581      * Retrieves the Smartthings API Subscription from a remote service, performing an API call
 582      *
 583      * @return stSub
 584      */
 585     public synchronized Optional<STSubscription> smartthingsSubscription() {
 586         if (apiKey.isBlank() || deviceId.isBlank()) {
 587             return Optional.empty();
 588         }
 589         Optional<STSubscription> stSub = Optional.empty();
 590         try {
 591             logger.info("{}: SSE Creating Smartthings Subscription", host);
 592             String contentString = new Gson().toJson(new JSONSubscriptionFilter(deviceId));
 593             logger.trace("{}: subscription: {}", host, contentString);
 594             InputStream subscriptionFilter = new ByteArrayInputStream(contentString.getBytes());
 595             URI uri = new URI("https", null, SMARTTHINGS_URL, 443, "/subscriptions", null, null);
 596             Optional<String> response = sendUrl(HttpMethod.POST, uri, subscriptionFilter);
 597             stSub = response.map(r -> new Gson().fromJson(r, STSubscription.class));
 598             if (!stSub.isPresent()) {
 599                 throw new IOException("No Data - is DeviceID correct?");
 600             }
 601         } catch (JsonSyntaxException | URISyntaxException | IOException e) {
 602             logger.warn("{}: SSE Subscription to Smartthings Cloud failed: {}", host, e.getMessage());
 603         }
 604         return stSub;
 605     }
 606
 607     public synchronized void startSSE() {
 608         if (!subscriptionRunning) {
 609             logger.trace("{}: SSE Starting job", host);
 610             subscription = smartthingsSubscription();
 611             logger.trace("{}: SSE got subscription ID: {}", host,
 612                     subscription.map(a -> a.getSubscriptionId()).orElse("None"));
 613             if (!subscription.map(a -> a.getSubscriptionId()).orElse("").isBlank()) {
 614                 receiveSSEEvents();
 615             }
 616         }
 617     }
 618
 619     public void stopSSE() {
 620         handlerWrapper.ifPresent(a -> {
 621             a.cancel();
 622             logger.trace("{}: SSE Stopping job", host);
 623             handlerWrapper = Optional.empty();
 624             subscriptionRunning = false;
 625         });
 626     }
 627
 628     /**
 629      * SubscriberWrapper needed to make async SSE stream cancelable
 630      *
 631      */
 632     @NonNullByDefault({})
 633     private static class SubscriberWrapper implements BodySubscriber<Void> {
 634         private final CountDownLatch latch;
 635         private final BodySubscriber<Void> subscriber;
 636         private Subscription subscription;
 637
 638         private SubscriberWrapper(BodySubscriber<Void> subscriber, CountDownLatch latch) {
 639             this.subscriber = subscriber;
 640             this.latch = latch;
 641         }
 642
 643         @Override
 644         public CompletionStage<Void> getBody() {
 645             return subscriber.getBody();
 646         }
 647
 648         @Override
 649         public void onSubscribe(Subscription subscription) {
 650             subscriber.onSubscribe(subscription);
 651             this.subscription = subscription;
 652             latch.countDown();
 653         }
 654
 655         @Override
 656         public void onNext(List<ByteBuffer> item) {
 657             subscriber.onNext(item);
 658         }
 659
 660         @Override
 661         public void onError(Throwable throwable) {
 662             subscriber.onError(throwable);
 663         }
 664
 665         @Override
 666         public void onComplete() {
 667             subscriber.onComplete();
 668         }
 669
 670         public void cancel() {
 671             subscription.cancel();
 672         }
 673     }
 674
 675     @NonNullByDefault({})
 676     private static class BodyHandlerWrapper implements BodyHandler<Void> {
 677         private final CountDownLatch latch = new CountDownLatch(1);
 678         private final BodyHandler<Void> handler;
 679         private SubscriberWrapper subscriberWrapper;
 680         private int statusCode = -1;
 681
 682         private BodyHandlerWrapper(BodyHandler<Void> handler) {
 683             this.handler = handler;
 684         }
 685
 686         @Override
 687         public BodySubscriber<Void> apply(ResponseInfo responseInfo) {
 688             subscriberWrapper = new SubscriberWrapper(handler.apply(responseInfo), latch);
 689             this.statusCode = responseInfo.statusCode();
 690             return subscriberWrapper;
 691         }
 692
 693         public void waitForEvent(boolean cancel) {
 694             try {
 695                 CompletableFuture.runAsync(() -> {
 696                     try {
 697                         latch.await();
 698                         if (cancel) {
 699                             subscriberWrapper.cancel();
 700                         }
 701                     } catch (InterruptedException ignore) {
 702                     }
 703                 }).get(2, TimeUnit.SECONDS);
 704             } catch (InterruptedException | ExecutionException | TimeoutException ignore) {
 705             }
 706         }
 707
 708         public int getStatusCode() {
 709             waitForEvent(false);
 710             return statusCode;
 711         }
 712
 713         public void cancel() {
 714             waitForEvent(true);
 715         }
 716     }
 717
 718     public void receiveSSEEvents() {
 719         subscription.ifPresent(sub -> {
 720             updateTV();
 721             try {
 722                 URI uri = new URI(sub.getregistrationUrl());
 723                 HttpClient client = HttpClient.newHttpClient();
 724                 HttpRequest request = HttpRequest.newBuilder(uri).timeout(Duration.ofSeconds(2)).GET()
 725                         .header("Authorization", "Bearer " + this.apiKey).build();
 726                 handlerWrapper = Optional.ofNullable(
 727                         new BodyHandlerWrapper(HttpResponse.BodyHandlers.ofByteArrayConsumer(b -> processSSEEvent(b))));
 728                 handlerWrapper.ifPresent(h -> {
 729                     client.sendAsync(request, h);
 730                 });
 731                 logger.debug("{}: SSE job {}", host, checkResponseCode() ? "Started" : "Failed");
 732             } catch (URISyntaxException e) {
 733                 logger.warn("{}: SSE URI Exception: {}", host, e.getMessage());
 734             }
 735         });
 736     }
 737
 738     boolean checkResponseCode() {
 739         int respCode = handlerWrapper.map(a -> a.getStatusCode()).orElse(-1);
 740         logger.trace("{}: SSE GOT Response Code: {}", host, respCode);
 741         subscriptionRunning = (respCode == 200);
 742         return subscriptionRunning;
 743     }
 744
 745     Map<String, String> bytesToMap(byte[] bytes) {
 746         String s = new String(bytes, StandardCharsets.UTF_8);
 747         // logger.trace("{}: SSE received: {}", host, s);
 748         Map<String, String> properties = new HashMap<String, String>();
 749         String[] pairs = s.split("\r?\n");
 750         for (String pair : pairs) {
 751             String[] kv = pair.split(":", 2);
 752             properties.put(kv[0].trim(), kv[1].trim());
 753         }
 754         logger.trace("{}: SSE received: {}", host, properties);
 755         updateTV();
 756         return properties;
 757     }
 758
 759     synchronized void processSSEEvent(Optional<byte[]> bytes) {
 760         bytes.ifPresent(b -> {
 761             Map<String, String> properties = bytesToMap(b);
 762             String rawData = properties.getOrDefault("data", "none");
 763             String event = properties.getOrDefault("event", "none");
 764             // logger.trace("{}: SSE Decoding event: {}", host, event);
 765             switch (event) {
 766                 case "CONTROL_EVENT":
 767                     subscriptionRunning = "welcome".equals(rawData);
 768                     if (!subscriptionRunning) {
 769                         logger.trace("{}: SSE Subscription ended", host);
 770                         startSSE();
 771                     }
 772                     break;
 773                 case "DEVICE_EVENT":
 774                     try {
 775                         // decode json here
 776                         Optional<STSSEData> data = Optional.ofNullable(new Gson().fromJson(rawData, STSSEData.class));
 777                         data.ifPresentOrElse(d -> {
 778                             d.setTvInfo(tvInfo);
 779                             String[] inputList = d.getInputSourceList();
 780                             if (inputList.length > 0) {
 781                                 logger.trace("{}: SSE Got input source list: {}", host, Arrays.asList(inputList));
 782                                 tvInfo.ifPresent(a -> a.updateSupportedInputSources(inputList));
 783                             }
 784                             String inputSource = d.getInputSource();
 785                             if (!inputSource.isBlank()) {
 786                                 updateState(SOURCE_NAME, inputSource);
 787                                 int sourceId = d.getInputSourceId();
 788                                 logger.trace("{}: SSE Got input source: {} ID: {}", host, inputSource, sourceId);
 789                                 updateState(SOURCE_ID, sourceId);
 790                             }
 791                             Number tvChannel = d.getTvChannel();
 792                             if (tvChannel.intValue() != -1) {
 793                                 updateState(CHANNEL, tvChannel);
 794                                 String tvChannelName = d.getTvChannelName();
 795                                 logger.trace("{}: SSE Got TV Channel Name: {} Channel: {}", host, tvChannelName,
 796                                         tvChannel);
 797                                 updateState(CHANNEL_NAME, tvChannelName);
 798                             }
 799                             String Power = d.getSwitch();
 800                             if (!Power.isBlank()) {
 801                                 logger.debug("{}: SSE Got TV Power: {}", host, Power);
 802                                 if ("on".equals(Power)) {
 803                                     // handler.putOnline(); // ignore on event for now
 804                                 } else {
 805                                     // handler.setOffline(); // ignore off event for now
 806                                 }
 807                             }
 808                         }, () -> logger.warn("{}: SSE Received NULL data", host));
 809                     } catch (JsonSyntaxException e) {
 810                         logger.warn("{}: SmartThingsApiService: Error ({}) in message: {}", host, e.getMessage(),
 811                                 rawData);
 812                     }
 813                     break;
 814                 default:
 815                     logger.trace("{}: SSE not handling event: {}", host, event);
 816                     break;
 817             }
 818         });
 819     }
 820
 821     private boolean updateDeviceID(TvValues.Items item) {
 822         this.deviceId = item.getDeviceId();
 823         logger.debug("{}: found {} device, adding device id {}", host, item.getName(), deviceId);
 824         handler.putConfig(SMARTTHINGS_DEVICEID, deviceId);
 825         prevUpdate = 0;
 826         return true;
 827     }
 828
 829     public boolean fetchdata() {
 830         if (System.currentTimeMillis() >= prevUpdate + RATE_LIMIT) {
 831             if (deviceId.isBlank()) {
 832                 tvInfo = fetchTVProperties(DEVICES);
 833                 boolean found = false;
 834                 if (tvInfo.isPresent()) {
 835                     TvValues t = tvInfo.get();
 836                     switch (t.getItems().length) {
 837                         case 0:
 838                         case 1:
 839                             logger.warn("{}: No devices found - please add your TV to the Smartthings app", host);
 840                             break;
 841                         case 2:
 842                             found = Arrays.asList(t.getItems()).stream().filter(a -> "Samsung TV".equals(a.getName()))
 843                                     .map(a -> updateDeviceID(a)).findFirst().orElse(false);
 844                             break;
 845                         default:
 846                             logger.warn("{}: No device Id selected, please enter one of the following:", host);
 847                             Arrays.asList(t.getItems()).stream().forEach(a -> logger.info("{}: '{}' : {}({})", host,
 848                                     a.getDeviceId(), a.getName(), a.getLabel()));
 849                     }
 850                 }
 851                 if (found) {
 852                     return fetchdata();
 853                 } else {
 854                     stop();
 855                     return false;
 856                 }
 857             }
 858             tvInfo = fetchTVProperties(COMPONENTS);
 859             prevUpdate = System.currentTimeMillis();
 860         }
 861         return (tvInfo.isPresent());
 862     }
 863
 864     @Override
 865     public void start() {
 866         online = true;
 867         errorCount = 0;
 868         startSSE();
 869     }
 870
 871     @Override
 872     public void stop() {
 873         online = false;
 874         stopSSE();
 875     }
 876
 877     @Override
 878     public void clearCache() {
 879         stateMap.clear();
 880         start();
 881     }
 882
 883     @Override
 884     public boolean isUpnp() {
 885         return false;
 886     }
 887
 888     @Override
 889     public boolean checkConnection() {
 890         return online;
 891     }
 892
 893     @Override
 894     public boolean handleCommand(String channel, Command command) {
 895         logger.trace("{}: Received channel: {}, command: {}", host, channel, command);
 896         if (!checkConnection()) {
 897             logger.trace("{}: Smartthings offline", host);
 898             return false;
 899         }
 900
 901         if (fetchdata()) {
 902             return tvInfo.map(t -> {
 903                 boolean result = false;
 904                 if (command == RefreshType.REFRESH) {
 905                     switch (channel) {
 906                         case CHANNEL_NAME:
 907                             updateState(CHANNEL_NAME, t.getTvChannelName());
 908                             break;
 909                         case CHANNEL:
 910                             updateState(CHANNEL, t.getTvChannel());
 911                             break;
 912                         case SOURCE_ID:
 913                         case SOURCE_NAME:
 914                             updateState(SOURCE_NAME, t.getInputSource());
 915                             updateState(SOURCE_ID, t.getInputSourceId());
 916                             break;
 917                         default:
 918                             break;
 919                     }
 920                     return true;
 921                 }
 922
 923                 switch (channel) {
 924                     case SOURCE_ID:
 925                         if (command instanceof DecimalType commandAsDecimalType) {
 926                             int val = commandAsDecimalType.intValue();
 927                             if (val >= 0 && val < t.getSources().length) {
 928                                 result = setSourceName(t.getSources()[val]);
 929                             } else {
 930                                 logger.warn("{}: Invalid source ID: {}, acceptable: 0..{}", host, command,
 931                                         t.getSources().length);
 932                             }
 933                         }
 934                         break;
 935                     case SOURCE_NAME:
 936                         if (command instanceof StringType) {
 937                             if (t.getSourcesString().contains(command.toString()) || t.getSourcesString().isBlank()) {
 938                                 result = setSourceName(command.toString());
 939                             } else {
 940                                 logger.warn("{}: Invalid source Name: {}, acceptable: {}", host, command,
 941                                         t.getSourcesString());
 942                             }
 943                         }
 944                         break;
 945                     default:
 946                         logger.warn("{}: Samsung TV doesn't support transmitting for channel '{}'", host, channel);
 947                 }
 948                 if (!result) {
 949                     logger.warn("{}: Smartthings: wrong command type {} channel {}", host, command, channel);
 950                 }
 951                 return result;
 952             }).orElse(false);
 953         }
 954         return false;
 955     }
 956
 957     private void updateState(String channel, Object value) {
 958         if (!stateMap.getOrDefault(channel, "None").equals(value)) {
 959             switch (channel) {
 960                 case CHANNEL:
 961                 case SOURCE_ID:
 962                     handler.valueReceived(channel, new DecimalType((Number) value));
 963                     break;
 964                 default:
 965                     handler.valueReceived(channel, new StringType((String) value));
 966                     break;
 967             }
 968             stateMap.put(channel, value);
 969         } else {
 970             logger.trace("{}: Value '{}' for {} hasn't changed, ignoring update", host, value, channel);
 971         }
 972     }
 973
 974     private boolean setSourceName(String value) {
 975         return setTVProperties("mediaInputSource", "setInputSource", value);
 976     }
 977 }