package com.oneandone.typedrest;

import java.net.*;
import rx.*;
import rx.functions.Action2;
import rx.schedulers.Schedulers;
import static rx.util.async.Async.runAsync;

/**
 * Base class for building REST endpoints that represents a stream of
 * <code>TEntity</code>s as <code>TElementEndpoint</code>s.
 *
 * @param <TEntity> The type of entity the endpoint represents.
 * @param <TElementEndpoint> The specific type of {@link ElementEndpoint} to
 * provide for individual <code>TEntity</code>s.
 */
public abstract class AbstractStreamEndpoint<TEntity, TElementEndpoint extends ElementEndpoint<TEntity>>
        extends AbstractPagedCollectionEndpoint<TEntity, TElementEndpoint> implements StreamEndpoint<TEntity, TElementEndpoint> {

    /**
     * Creates a new element stream endpoint.
     *
     * @param parent The parent endpoint containing this one.
     * @param relativeUri The URI of this endpoint relative to the
     * <code>parent</code>'s. Missing trailing slash will be appended
     * automatically.
     * @param entityType The type of entity the endpoint represents.
     */
    protected AbstractStreamEndpoint(Endpoint parent, URI relativeUri, Class<TEntity> entityType) {
        super(parent, relativeUri, entityType);
    }

    /**
     * Creates a new element stream endpoint.
     *
     * @param parent The parent endpoint containing this one.
     * @param relativeUri The URI of this endpoint relative to the
     * <code>parent</code>'s. Missing trailing slash will be appended
     * automatically.
     * @param entityType The type of entity the endpoint represents.
     */
    protected AbstractStreamEndpoint(Endpoint parent, String relativeUri, Class<TEntity> entityType) {
        super(parent, relativeUri, entityType);
    }

    @Override
    public rx.Observable<TEntity> getObservable() {
        return getObservable(0);
    }

    @Override
    public rx.Observable<TEntity> getObservable(long startIndex) {
        return getObservable(startIndex, Schedulers.io());
    }

    /**
     * Provides an observable stream of elements.
     *
     * @param startIndex The index of the first element to return in the stream.
     * Use negative values to start counting from the end of the stream.
     * @param scheduler The scheduler used to run the background thread.
     * @return An observable stream of elements.
     */
    rx.Observable<TEntity> getObservable(final long startIndex, Scheduler scheduler) {
        return runAsync(scheduler, new Action2<rx.Observer<? super TEntity>, Subscription>() {

            @Override
            public void call(rx.Observer<? super TEntity> observer, Subscription subscription) {
                long currentStartIndex = startIndex;
                while (!subscription.isUnsubscribed()) {
                    PartialResponse<TEntity> response;
                    try {
                        response = (currentStartIndex >= 0)
                                ? readRange(currentStartIndex, null)
                                : readRange(null, -currentStartIndex);
                    } catch (IndexOutOfBoundsException ex) {
                        // No new data available yet, keep polling
                        continue;
                    } catch (Throwable error) {
                        observer.onError(error);
                        return;
                    }

                    for (TEntity entity : response.getElements()) {
                        observer.onNext(entity);
                    }

                    if (response.isEndReached()) {
                        observer.onCompleted();
                        return;
                    }

                    // Continue polling for more data
                    currentStartIndex = response.getTo() + 1;
                }
            }
        });
    }
}

@@ -63,37 +63,8 @@
 .*;%0A
-import rx.functions.Action2;%0A
 impo
@@ -2463,119 +2463,8 @@
 er, 
-new Action2%3Crx.Observer%3C? super TEntity%3E, Subscription%3E() %7B%0A%0A            @Override%0A            public void call
 (rx.
@@ -2526,22 +2526,21 @@
 iption) 
-%7B%0A    
+-%3E %7B%0A
         
@@ -2592,20 +2592,16 @@
         
-    
 while (!
@@ -2649,20 +2649,16 @@
         
-    
 PartialR
@@ -2704,22 +2704,14 @@
     
-  
-  
 try %7B%0A
-    
     
@@ -2762,20 +2762,16 @@
 x %3E= 0)%0A
-    
         
@@ -2855,20 +2855,16 @@
         
-    
 : readRa
@@ -2894,20 +2894,16 @@
 Index);%0A
-    
         
@@ -2971,20 +2971,16 @@
         
-  
-  
 // No ne
@@ -3034,20 +3034,16 @@
         
-    
 continue
@@ -3052,36 +3052,32 @@
                 
-   
- 
 %7D catch (Throwab
@@ -3084,28 +3084,24 @@
 le error) %7B%0A
-    
             
@@ -3145,36 +3145,32 @@
                 
-    
 return;%0A        
@@ -3169,36 +3169,32 @@
                 
- 
-   
 %7D%0A%0A             
@@ -3200,92 +3200,56 @@
     
-    for (TEntity entity : response.getElements()) %7B%0A                        
+response.getElements().stream().forEach(
 observer
 .onN
@@ -3248,51 +3248,20 @@
 rver
-.
+::
 onNext
-(entity);%0A                    %7D%0A%0A   
+);%0A%0A
     
@@ -3272,17 +3272,16 @@
         
- 
 if (resp
@@ -3295,36 +3295,32 @@
 EndReached()) %7B%0A
-    
                 
@@ -3359,36 +3359,32 @@
                 
-    
 return;%0A        
@@ -3387,35 +3387,27 @@
             
- 
-   %7D%0A%0A    
+%7D%0A%0A
             
@@ -3460,20 +3460,16 @@
         
-   
- 
 currentS
@@ -3502,34 +3502,16 @@
 () + 1;%0A
-                %7D%0A
         

Use lambdas