From 7958827a0ba1126646e01314e7c6bb4f86292dc8 Mon Sep 17 00:00:00 2001
From: Azul <azul@riseup.net>
Date: Tue, 10 Sep 2013 18:57:52 +0200
Subject: use our own JsonStream and a created callback in CouchChanges

---
 lib/tapicero.rb                |  1 +
 lib/tapicero/config.rb         |  2 +-
 lib/tapicero/couch_changes.rb  | 20 ++++++++++++++++++--
 lib/tapicero/couch_stream.rb   |  4 +---
 lib/tapicero/json_stream.rb    | 28 ++++++++++++++++++++++++++++
 lib/tapicero_daemon.rb         | 17 ++++++-----------
 test/unit/couch_stream_test.rb |  8 +-------
 7 files changed, 56 insertions(+), 24 deletions(-)
 create mode 100644 lib/tapicero/json_stream.rb

diff --git a/lib/tapicero.rb b/lib/tapicero.rb
index ee52db4..0d49b38 100644
--- a/lib/tapicero.rb
+++ b/lib/tapicero.rb
@@ -15,5 +15,6 @@ Tapicero::Config.load(BASE_DIR, 'config/default.yaml', LEAP_CA_CONFIG, ARGV.grep
 #
 # Load Tapicero
 #
+require 'tapicero/json_stream'
 require 'tapicero/couch_stream'
 require 'tapicero/couch_changes'
diff --git a/lib/tapicero/config.rb b/lib/tapicero/config.rb
index cb22792..ab00460 100644
--- a/lib/tapicero/config.rb
+++ b/lib/tapicero/config.rb
@@ -18,7 +18,7 @@ module Tapicero
     def couch_host
       couch_connection[:protocol] + '://' +
         couch_connection[:host] + ':' +
-        couch_connection[:port] + '/'
+        couch_connection[:port].to_s + '/'
     end
 
     private
diff --git a/lib/tapicero/couch_changes.rb b/lib/tapicero/couch_changes.rb
index 2e668dc..abb451d 100644
--- a/lib/tapicero/couch_changes.rb
+++ b/lib/tapicero/couch_changes.rb
@@ -4,16 +4,32 @@ module Tapicero
       @stream = stream
     end
 
+    def created(hash = {}, &block)
+      if block_given?
+        @created = block
+      else
+        @created && @created.call(hash)
+      end
+    end
+
     def last_seq
       @stream.get "_changes", :limit => 1, :descending => true do |hash|
         return hash[:last_seq]
       end
     end
 
-    def follow
+    def listen
       @stream.get "_changes", :feed => :continuous, :since => last_seq do |hash|
-        yield(hash)
+        callbacks(hash)
       end
     end
+
+    def callbacks(hash)
+      #changed
+      return if hash[:deleted]
+      return unless changes = hash[:changes]
+      return created(hash) if changes[0][:rev].start_with?('1-')
+      #updated
+    end
   end
 end
diff --git a/lib/tapicero/couch_stream.rb b/lib/tapicero/couch_stream.rb
index 96c752a..8583010 100644
--- a/lib/tapicero/couch_stream.rb
+++ b/lib/tapicero/couch_stream.rb
@@ -1,5 +1,3 @@
-require 'yajl/http_stream'
-
 module Tapicero
   class CouchStream
     def initialize(database_url)
@@ -9,7 +7,7 @@ module Tapicero
     def get(path, options)
       url = url_for(path, options)
       # puts url
-      Yajl::HttpStream.get(url, :symbolize_keys => true) do |hash|
+      Tapicero::JsonStream.get(url, :symbolize_keys => true) do |hash|
         yield(hash)
       end
     end
diff --git a/lib/tapicero/json_stream.rb b/lib/tapicero/json_stream.rb
new file mode 100644
index 0000000..d906ce2
--- /dev/null
+++ b/lib/tapicero/json_stream.rb
@@ -0,0 +1,28 @@
+require 'net/http'
+require 'uri'
+require 'yajl'
+
+
+# Since Yajl HTTP Stream will go a way in version 2.0 here's a simple substitude.
+module Tapicero
+  class JsonStream
+
+    def self.get(url, options, &block)
+      uri = URI(url)
+      parser = Yajl::Parser.new(options)
+      parser.on_parse_complete = block
+      Net::HTTP.start(uri.host, uri.port) do |http|
+        request = Net::HTTP::Get.new uri.request_uri
+
+
+        http.request request do |response|
+          response.read_body do |chunk|
+            parser << chunk
+          end
+        end
+      end
+
+    end
+  end
+end
+
diff --git a/lib/tapicero_daemon.rb b/lib/tapicero_daemon.rb
index a5347c7..5f754c7 100644
--- a/lib/tapicero_daemon.rb
+++ b/lib/tapicero_daemon.rb
@@ -8,18 +8,13 @@
 require 'tapicero'
 
 module Tapicero
+  puts    " * Observing #{Config.couch_host}"
   puts    " * Tracking #{Config.users_db_name}"
   couch   = CouchStream.new(Config.couch_host + Config.users_db_name)
-  changes = CouchChanges.new(couch)
-
-  # fill the pool
-  # pool.fill
-
-  # watch for deletions, fill the pool whenever it gets low
-  changes.follow do |hash|
-    if hash[:created]
-      puts " Created #{hash.inspect}"
-      # pool.fill
-    end
+  users = CouchChanges.new(couch)
+  users.created do |hash|
+    puts "Created user " + hash[:id]
   end
+
+  users.listen
 end
diff --git a/test/unit/couch_stream_test.rb b/test/unit/couch_stream_test.rb
index 48b663a..a9de21f 100644
--- a/test/unit/couch_stream_test.rb
+++ b/test/unit/couch_stream_test.rb
@@ -1,12 +1,6 @@
 require File.expand_path('../../test_helper.rb', __FILE__)
 require 'tapicero/couch_stream'
 
-# we'll mock this
-module Yajl
-  class HttpStream
-  end
-end
-
 class CouchStreamTest < MiniTest::Unit::TestCase
 
   def setup
@@ -18,7 +12,7 @@ class CouchStreamTest < MiniTest::Unit::TestCase
   end
 
   def test_get
-    Yajl::HttpStream.expects(:get).
+    Tapicero::JsonStream.expects(:get).
       with(@url, :symbolize_keys => true).
       yields(stub_hash = stub)
     @stream.get(@path, @options) do |hash|
-- 
cgit v1.2.3