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

   1 /**
   2  * Copyright (c) 2010-2021 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.nest.internal.sdm.api;
  14
  15 import static org.eclipse.jetty.http.HttpHeader.*;
  16 import static org.eclipse.jetty.http.HttpMethod.POST;
  17 import static org.openhab.binding.nest.internal.sdm.dto.SDMGson.GSON;
  18
  19 import java.io.IOException;
  20 import java.time.Duration;
  21 import java.util.HashMap;
  22 import java.util.HashSet;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Set;
  26 import java.util.concurrent.ExecutionException;
  27 import java.util.concurrent.ScheduledThreadPoolExecutor;
  28 import java.util.concurrent.TimeUnit;
  29 import java.util.concurrent.TimeoutException;
  30 import java.util.function.Consumer;
  31 import java.util.stream.Collectors;
  32
  33 import org.eclipse.jdt.annotation.NonNullByDefault;
  34 import org.eclipse.jetty.client.HttpClient;
  35 import org.eclipse.jetty.client.util.StringContentProvider;
  36 import org.eclipse.jetty.http.HttpMethod;
  37 import org.openhab.binding.nest.internal.sdm.dto.PubSubRequestsResponses.PubSubAcknowledgeRequest;
  38 import org.openhab.binding.nest.internal.sdm.dto.PubSubRequestsResponses.PubSubCreateRequest;
  39 import org.openhab.binding.nest.internal.sdm.dto.PubSubRequestsResponses.PubSubPullRequest;
  40 import org.openhab.binding.nest.internal.sdm.dto.PubSubRequestsResponses.PubSubPullResponse;
  41 import org.openhab.binding.nest.internal.sdm.exception.FailedSendingPubSubDataException;
  42 import org.openhab.binding.nest.internal.sdm.exception.InvalidPubSubAccessTokenException;
  43 import org.openhab.binding.nest.internal.sdm.exception.InvalidPubSubAuthorizationCodeException;
  44 import org.openhab.binding.nest.internal.sdm.listener.PubSubSubscriptionListener;
  45 import org.openhab.core.auth.client.oauth2.AccessTokenResponse;
  46 import org.openhab.core.auth.client.oauth2.OAuthClientService;
  47 import org.openhab.core.auth.client.oauth2.OAuthException;
  48 import org.openhab.core.auth.client.oauth2.OAuthFactory;
  49 import org.openhab.core.auth.client.oauth2.OAuthResponseException;
  50 import org.openhab.core.common.NamedThreadFactory;
  51 import org.openhab.core.io.net.http.HttpClientFactory;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54
  55 /**
  56  * The {@link PubSubAPI} implements a subset of the Pub/Sub REST API which allows for subscribing to SDM events.
  57  *
  58  * @author Wouter Born - Initial contribution
  59  *
  60  * @see https://cloud.google.com/pubsub/docs/reference/rest
  61  * @see https://developers.google.com/nest/device-access/api/events
  62  */
  63 @NonNullByDefault
  64 public class PubSubAPI {
  65
  66     private class Subscriber implements Runnable {
  67
  68         private final String subscriptionId;
  69
  70         Subscriber(String subscriptionId) {
  71             this.subscriptionId = subscriptionId;
  72         }
  73
  74         @Override
  75         public void run() {
  76             if (!subscriptionListeners.containsKey(subscriptionId)) {
  77                 logger.debug("Stop receiving subscription '{}' messages since there are no listeners", subscriptionId);
  78                 return;
  79             }
  80
  81             try {
  82                 String messages = pullSubscriptionMessages(subscriptionId);
  83
  84                 PubSubPullResponse pullResponse = GSON.fromJson(messages, PubSubPullResponse.class);
  85
  86                 if (pullResponse != null && pullResponse.receivedMessages != null) {
  87                     logger.debug("Subscription '{}' has {} new message(s)", subscriptionId,
  88                             pullResponse.receivedMessages.size());
  89                     forEachListener((listener) -> pullResponse.receivedMessages
  90                             .forEach((message) -> listener.onMessage(message.message)));
  91                     List<String> ackIds = pullResponse.receivedMessages.stream().map(message -> message.ackId)
  92                             .collect(Collectors.toList());
  93                     acknowledgeSubscriptionMessages(subscriptionId, ackIds);
  94                 } else {
  95                     forEachListener((listener) -> listener.onNoNewMessages());
  96                 }
  97
  98                 scheduler.submit(this);
  99             } catch (FailedSendingPubSubDataException e) {
 100                 logger.debug("Expected exception while pulling message for '{}' subscription", subscriptionId, e);
 101                 Throwable cause = e.getCause();
 102                 if (!(cause instanceof InterruptedException)) {
 103                     forEachListener((listener) -> listener.onError(e));
 104                     scheduler.schedule(this, RETRY_TIMEOUT.toNanos(), TimeUnit.NANOSECONDS);
 105                 }
 106             } catch (InvalidPubSubAccessTokenException e) {
 107                 logger.warn("Cannot pull messages for '{}' subscription (access token invalid)", subscriptionId, e);
 108                 forEachListener((listener) -> listener.onError(e));
 109             } catch (Exception e) {
 110                 logger.warn("Unexpected exception while pulling message for '{}' subscription", subscriptionId, e);
 111                 forEachListener((listener) -> listener.onError(e));
 112                 scheduler.schedule(this, RETRY_TIMEOUT.toNanos(), TimeUnit.NANOSECONDS);
 113             }
 114         }
 115
 116         private void forEachListener(Consumer<PubSubSubscriptionListener> consumer) {
 117             Set<PubSubSubscriptionListener> listeners = subscriptionListeners.get(subscriptionId);
 118             if (listeners != null) {
 119                 listeners.forEach(consumer::accept);
 120             } else {
 121                 logger.debug("Subscription '{}' has no listeners", subscriptionId);
 122             }
 123         }
 124     }
 125
 126     private static final String AUTH_URL = "https://accounts.google.com/o/oauth2/auth";
 127     private static final String TOKEN_URL = "https://accounts.google.com/o/oauth2/token";
 128     private static final String REDIRECT_URI = "urn:ietf:wg:oauth:2.0:oob";
 129
 130     private static final String PUBSUB_HANDLE_FORMAT = "%s.pubsub";
 131     private static final String PUBSUB_SCOPE = "https://www.googleapis.com/auth/pubsub";
 132
 133     private static final String PUBSUB_URL_PREFIX = "https://pubsub.googleapis.com/v1/";
 134     private static final int PUBSUB_PULL_MAX_MESSAGES = 10;
 135
 136     private static final String APPLICATION_JSON = "application/json";
 137     private static final String BEARER = "Bearer ";
 138
 139     private static final Duration REQUEST_TIMEOUT = Duration.ofMinutes(1);
 140     private static final Duration RETRY_TIMEOUT = Duration.ofSeconds(30);
 141
 142     private final Logger logger = LoggerFactory.getLogger(PubSubAPI.class);
 143
 144     private final HttpClient httpClient;
 145     private final OAuthClientService oAuthService;
 146     private final String projectId;
 147     private final ScheduledThreadPoolExecutor scheduler;
 148     private final Map<String, Set<PubSubSubscriptionListener>> subscriptionListeners = new HashMap<>();
 149
 150     public PubSubAPI(HttpClientFactory httpClientFactory, OAuthFactory oAuthFactory, String ownerId, String projectId,
 151             String clientId, String clientSecret) {
 152         this.httpClient = httpClientFactory.getCommonHttpClient();
 153         this.projectId = projectId;
 154         this.oAuthService = oAuthFactory.createOAuthClientService(String.format(PUBSUB_HANDLE_FORMAT, ownerId),
 155                 TOKEN_URL, AUTH_URL, clientId, clientSecret, PUBSUB_SCOPE, false);
 156         scheduler = new ScheduledThreadPoolExecutor(3, new NamedThreadFactory(ownerId, true));
 157     }
 158
 159     public void dispose() {
 160         subscriptionListeners.clear();
 161         scheduler.shutdownNow();
 162     }
 163
 164     public void authorizeClient(String authorizationCode) throws InvalidPubSubAuthorizationCodeException, IOException {
 165         try {
 166             oAuthService.getAccessTokenResponseByAuthorizationCode(authorizationCode, REDIRECT_URI);
 167         } catch (OAuthException | OAuthResponseException e) {
 168             throw new InvalidPubSubAuthorizationCodeException(
 169                     "Failed to authorize Pub/Sub client. Check the authorization code or generate a new one.", e);
 170         }
 171     }
 172
 173     public void checkAccessTokenValidity() throws InvalidPubSubAccessTokenException, IOException {
 174         getAuthorizationHeader();
 175     }
 176
 177     private String acknowledgeSubscriptionMessages(String subscriptionId, List<String> ackIds)
 178             throws FailedSendingPubSubDataException, InvalidPubSubAccessTokenException {
 179         logger.debug("Acknowleding {} message(s) for '{}' subscription", ackIds.size(), subscriptionId);
 180         String url = getSubscriptionUrl(subscriptionId) + ":acknowledge";
 181         String requestContent = GSON.toJson(new PubSubAcknowledgeRequest(ackIds));
 182         return postJson(url, requestContent);
 183     }
 184
 185     public void addSubscriptionListener(String subscriptionId, PubSubSubscriptionListener listener) {
 186         synchronized (subscriptionListeners) {
 187             Set<PubSubSubscriptionListener> listeners = subscriptionListeners.get(subscriptionId);
 188             if (listeners == null) {
 189                 listeners = new HashSet<>();
 190                 subscriptionListeners.put(subscriptionId, listeners);
 191             }
 192             listeners.add(listener);
 193             if (listeners.size() == 1) {
 194                 scheduler.submit(new Subscriber(subscriptionId));
 195             }
 196         }
 197     }
 198
 199     public void removeSubscriptionListener(String subscriptionId, PubSubSubscriptionListener listener) {
 200         synchronized (subscriptionListeners) {
 201             Set<PubSubSubscriptionListener> listeners = subscriptionListeners.get(subscriptionId);
 202             if (listeners != null) {
 203                 listeners.remove(listener);
 204                 if (listeners.isEmpty()) {
 205                     subscriptionListeners.remove(subscriptionId);
 206                     scheduler.getQueue().removeIf((runnable) -> runnable instanceof Subscriber
 207                             && ((Subscriber) runnable).subscriptionId.equals(subscriptionId));
 208                 }
 209             }
 210         }
 211     }
 212
 213     public void createSubscription(String subscriptionId, String topicName)
 214             throws FailedSendingPubSubDataException, InvalidPubSubAccessTokenException {
 215         logger.debug("Creating '{}' subscription", subscriptionId);
 216         String url = getSubscriptionUrl(subscriptionId);
 217         String requestContent = GSON.toJson(new PubSubCreateRequest(topicName, true));
 218         putJson(url, requestContent);
 219     }
 220
 221     private String getAuthorizationHeader() throws InvalidPubSubAccessTokenException, IOException {
 222         try {
 223             AccessTokenResponse response = oAuthService.getAccessTokenResponse();
 224             if (response == null || response.getAccessToken() == null || response.getAccessToken().isEmpty()) {
 225                 throw new InvalidPubSubAccessTokenException(
 226                         "No Pub/Sub access token. Client may not have been authorized.");
 227             }
 228             return BEARER + response.getAccessToken();
 229         } catch (OAuthException | OAuthResponseException e) {
 230             throw new InvalidPubSubAccessTokenException(
 231                     "Error fetching Pub/Sub access token. Check the authorization code or generate a new one.", e);
 232         }
 233     }
 234
 235     private String getSubscriptionUrl(String subscriptionId) {
 236         return PUBSUB_URL_PREFIX + "projects/" + projectId + "/subscriptions/" + subscriptionId;
 237     }
 238
 239     private String postJson(String url, String requestContent)
 240             throws FailedSendingPubSubDataException, InvalidPubSubAccessTokenException {
 241         try {
 242             logger.debug("Posting JSON to: {}", url);
 243             String response = httpClient.newRequest(url) //
 244                     .method(POST) //
 245                     .header(ACCEPT, APPLICATION_JSON) //
 246                     .header(AUTHORIZATION, getAuthorizationHeader()) //
 247                     .content(new StringContentProvider(requestContent), APPLICATION_JSON) //
 248                     .timeout(REQUEST_TIMEOUT.toNanos(), TimeUnit.NANOSECONDS) //
 249                     .send() //
 250                     .getContentAsString();
 251             logger.debug("Response: {}", response);
 252             return response;
 253         } catch (ExecutionException | InterruptedException | IOException | TimeoutException e) {
 254             throw new FailedSendingPubSubDataException("Failed to send JSON POST request", e);
 255         }
 256     }
 257
 258     private String pullSubscriptionMessages(String subscriptionId)
 259             throws FailedSendingPubSubDataException, InvalidPubSubAccessTokenException {
 260         logger.debug("Pulling messages for '{}' subscription", subscriptionId);
 261         String url = getSubscriptionUrl(subscriptionId) + ":pull";
 262         String requestContent = GSON.toJson(new PubSubPullRequest(PUBSUB_PULL_MAX_MESSAGES));
 263         return postJson(url, requestContent);
 264     }
 265
 266     private String putJson(String url, String requestContent)
 267             throws FailedSendingPubSubDataException, InvalidPubSubAccessTokenException {
 268         try {
 269             logger.debug("Putting JSON to: {}", url);
 270             String response = httpClient.newRequest(url) //
 271                     .method(HttpMethod.PUT) //
 272                     .header(ACCEPT, APPLICATION_JSON) //
 273                     .header(AUTHORIZATION, getAuthorizationHeader()) //
 274                     .content(new StringContentProvider(requestContent), APPLICATION_JSON) //
 275                     .timeout(REQUEST_TIMEOUT.toNanos(), TimeUnit.NANOSECONDS) //
 276                     .send() //
 277                     .getContentAsString();
 278             logger.debug("Response: {}", response);
 279             return response;
 280         } catch (ExecutionException | InterruptedException | IOException | TimeoutException e) {
 281             throw new FailedSendingPubSubDataException("Failed to send JSON PUT request", e);
 282         }
 283     }
 284 }