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

   1 /**
   2  * Copyright (c) 2010-2023 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.persistence.dynamodb.internal;
  14
  15 import java.util.ArrayList;
  16 import java.util.List;
  17 import java.util.concurrent.CompletableFuture;
  18 import java.util.concurrent.atomic.AtomicInteger;
  19
  20 import org.eclipse.jdt.annotation.NonNullByDefault;
  21 import org.eclipse.jdt.annotation.Nullable;
  22 import org.reactivestreams.Subscriber;
  23 import org.reactivestreams.Subscription;
  24
  25 /**
  26  * Subscriber that subscribes the page of interest
  27  *
  28  * @author Sami Salonen - Initial contribution
  29  */
  30 @NonNullByDefault
  31 public class PageOfInterestSubscriber<T> implements Subscriber<T> {
  32
  33     private AtomicInteger skipped = new AtomicInteger();
  34     private int skip;
  35     private @Nullable Subscription subscription;
  36     private int pageIndex;
  37     private int pageSize;
  38     private List<T> page;
  39     private CompletableFuture<List<T>> future;
  40
  41     /**
  42      * Create new PageOfInterestSubscriber
  43      *
  44      * @param subscriber subscriber to get the page of interest
  45      * @param pageIndex page index that we want subscribe
  46      * @param pageSize page size
  47      */
  48     protected PageOfInterestSubscriber(CompletableFuture<List<T>> future, int pageIndex, int pageSize) {
  49         this.future = future;
  50         this.pageIndex = pageIndex;
  51         this.pageSize = pageSize;
  52         this.page = new ArrayList<>();
  53         this.skip = pageIndex * pageSize;
  54     }
  55
  56     @Override
  57     public void onSubscribe(@Nullable Subscription subscription) {
  58         this.subscription = subscription;
  59         if (subscription != null) {
  60             subscription.request(pageSize * (pageIndex + 1));
  61         }
  62     }
  63
  64     @Override
  65     public void onNext(T t) {
  66         Subscription localSubscription = subscription;
  67         if (localSubscription == null) {
  68             throw new IllegalStateException(
  69                     "Subscriber API has been contract violated: expecting a non-null subscriber");
  70         }
  71         if (future.isCancelled()) {
  72             localSubscription.cancel();
  73             onError(new InterruptedException());
  74         } else if (skipped.getAndIncrement() >= skip && page.size() < pageSize) {
  75             // We have skipped enough, start accumulating
  76             page.add(t);
  77             if (page.size() == pageSize) {
  78                 // We have the full page read
  79                 localSubscription.cancel();
  80                 onComplete();
  81             }
  82         }
  83     }
  84
  85     @Override
  86     public void onError(@NonNullByDefault({}) Throwable t) {
  87         if (!future.isDone()) {
  88             future.completeExceptionally(t);
  89         }
  90     }
  91
  92     @Override
  93     public void onComplete() {
  94         if (!future.isDone()) {
  95             future.complete(page);
  96         }
  97     }
  98 }