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