1 /*
   2  * Copyright (c) 2016, Oracle and/or its affiliates. All rights reserved.
   3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4  *
   5  * This code is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License version 2 only, as
   7  * published by the Free Software Foundation.  Oracle designates this
   8  * particular file as subject to the "Classpath" exception as provided
   9  * by Oracle in the LICENSE file that accompanied this code.
  10  *
  11  * This code is distributed in the hope that it will be useful, but WITHOUT
  12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14  * version 2 for more details (a copy is included in the LICENSE file that
  15  * accompanied this code).
  16  *
  17  * You should have received a copy of the GNU General Public License version
  18  * 2 along with this work; if not, write to the Free Software Foundation,
  19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  20  *
  21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  22  * or visit www.oracle.com if you need additional information or have any
  23  * questions.
  24  */
  25 
  26 package jdk.incubator.http;
  27 
  28 import java.util.Iterator;
  29 import java.util.concurrent.Flow;
  30 
  31 /**
  32  * A Publisher that is expected to run in same thread as subscriber.
  33  * Items are obtained from Iterable. Each new subscription gets a new Iterator.
  34  */
  35 class PullPublisher<T> implements Flow.Publisher<T> {
  36 
  37     private final Iterable<T> iterable;
  38 
  39     PullPublisher(Iterable<T> iterable) {
  40         this.iterable = iterable;
  41     }
  42 
  43     @Override
  44     public void subscribe(Flow.Subscriber<? super T> subscriber) {
  45         subscriber.onSubscribe(new Subscription(subscriber, iterable.iterator()));
  46     }
  47 
  48     private class Subscription implements Flow.Subscription {
  49 
  50         private final Flow.Subscriber<? super T> subscriber;
  51         private final Iterator<T> iter;
  52         private boolean done = false;
  53         private long demand = 0;
  54         private int recursion = 0;
  55 
  56         Subscription(Flow.Subscriber<? super T> subscriber, Iterator<T> iter) {
  57             this.subscriber = subscriber;
  58             this.iter = iter;
  59         }
  60 
  61         @Override
  62         public void request(long n) {
  63             if (done) {
  64                 subscriber.onError(new IllegalArgumentException("request(" + n + ")"));
  65             }
  66             demand += n;
  67             recursion ++;
  68             if (recursion > 1) {
  69                 return;
  70             }
  71             while (demand > 0) {
  72                 done = !iter.hasNext();
  73                 if (done) {
  74                     subscriber.onComplete();
  75                     recursion --;
  76                     return;
  77                 }
  78                 subscriber.onNext(iter.next());
  79                 demand --;
  80             }
  81         }
  82 
  83         @Override
  84         public void cancel() {
  85             done = true;
  86         }
  87 
  88     }
  89 }