benjchristensen

  @Test def runTests() {
    // using closure
    Observable.toObservable("1", "2", "3")
      .take(2)
      .subscribe((callback: String) => {
          println(callback)
      })
      
    // using Map of closures
    Observable.toObservable("1", "2", "3")

benjchristensen

/**
 * Demonstrate how Rx is used to compose Observables together
 * such as how a web service would to generate a JSON response.
 * 
 * The simulated methods for the metadata represent different 
 * services that are often backed by network calls.
 * 
 * This will return a sequence of dictionaries such as this:
 * 
 *  [id:1000, title:video-1000-title, length:5428, bookmark:0, 

benjchristensen

package rx.examples.groovy;

import rx.observables.Observable
import rx.observables.Observer;
import rx.observables.Subscription;
import rx.util.functions.Func1;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

benjchristensen

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class CallbackB {

    /**

benjchristensen

/**
 * Asynchronously calls 'customObservableNonBlocking' and defines 
 * a chain of operators to apply to the callback sequence.
 */
def simpleComposition() {
  // fetch an asynchronous Observable<String> 
  // that emits 75 Strings of 'anotherValue_#'
  customObservableNonBlocking()
    // skip the first 10
    .skip(10)

benjchristensen

(defn simpleComposition []
  "Asynchronously calls 'customObservableNonBlocking' and defines a 
   chain of operators to apply to the callback sequence."
  (-> 
    ; fetch an asynchronous Observable<String> 
    ; that emits 75 Strings of 'anotherValue_#'
    (customObservableNonBlocking)
    ; skip the first 10
    (.skip 10)
    ; take the next 5

benjchristensen

/**
 * Non-blocking method that immediately returns the value
 * if available or  uses a thread to fetch the value and 
 * callback via `onNext()` when done. 
 */
def Observable<T> getData(int id) {
    if(availableInMemory) {
        // if data available return immediately with data
        return Observable.create({ observer ->
            observer.onNext(valueFromMemory);

benjchristensen

ExecutorService executor = new ThreadPoolExecutor(4, 4, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>());

// get f3 with dependent result from f1
Future<String> f1 = executor.submit(new CallToRemoteServiceA());
Future<String> f3 = executor.submit(new CallToRemoteServiceC(f1.get()));

/* The work below can not proceed until f1.get()
   completes even though there is no dependency */

// also get f4/f5 after dependency f2 completes 

benjchristensen

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

benjchristensen

ExecutorService executor = new ThreadPoolExecutor(4, 4, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>());

Future<String> f1 = executor.submit(new CallToRemoteServiceA());
Future<String> f2 = executor.submit(new CallToRemoteServiceB());

System.out.println(f1.get() + " - " + f2.get());