using_streams.dart

library streams;

import 'dart:html';
import 'dart:async';
import 'dart:uri';
import 'dart:json' as JSON;
import 'package:js/js.dart' as js;

main() {
  var inputs =
    query("#query").
    onKeyUp.
    map((event) => event.target.value).
    where((text) => text.length > 2).
    transform(new Throttle(500)).
    distinct();

  inputs.
    map(queryWikipedia).
    transform(new Latest()).
    map((json) => json[1]).
    where((list) => ! list.isEmpty).
    listen(updateResults);
}

updateResults(items){
  var ul = query("#items");
  ul.innerHtml = items.map((_) => "<li>${_}</li>").join("");
}

class Throttle<T> extends StreamEventTransformer<T,T> {
  final duration;
  Timer lastTimer;
  Throttle(millis) :duration = new Duration(milliseconds : millis);

  void handleData(T event, EventSink sink) {
    if (lastTimer != null) {
      lastTimer.cancel();
    }
    lastTimer = new Timer(duration, () => sink.add(event));
  }
}

class Latest<T> extends StreamEventTransformer<Stream<T>, T> {
  StreamSubscription latest;

  void handleData(Stream<T> stream, EventSink<T> sink) {
    if(latest != null){
      latest.cancel();
    }
    latest = stream.listen(sink.add);
  }
}

Stream<Map> queryWikipedia(String term) {
  var c = new StreamController();

  js.scoped(() {
    js.context.wikipediaCallback = new js.Callback.once((jsonData) {
      var n = JSON.parse(js.context.JSON.stringify(jsonData));
      c.add(n);
    });

    var script = new Element.tag("script");
    script.src = "http://en.wikipedia.org/w/api.php?action=opensearch&search=${encodeUriComponent(term)}&callback=wikipediaCallback";
    document.body.children.add(script);
  });

  return c.stream;
}