journal: more postgres tests

2 files changed, 65 insertions, 12 deletions

diff --git a/lib/rbot/journal.rb b/lib/rbot/journal.rb
index cc0578de..71b0c7ff 100644
--- a/lib/rbot/journal.rb
+++ b/lib/rbot/journal.rb
@@ -76,7 +76,7 @@ module Journal
     end
 
     def ==(other)
-      @id == other.id
+      (@id == other.id) rescue false
     end
 
     def self.create(topic, payload, opt={})
@@ -104,16 +104,32 @@ module Journal
       end
 
       # returns a array of message instances that match the query
-      def find(query, limit=10, offset=0)
+      def find(query=nil, limit=100, offset=0)
       end
 
       # returns the number of messages that match the query
-      def count(query)
+      def count(query=nil)
       end
 
-      # delete messages that match the query
-      def delete(query)
+      # remove messages that match the query
+      def remove(query=nil)
       end
+
+      # destroy the underlying table/collection
+      def drop
+      end
+
+      # Returns all classes from the namespace that implement this interface
+      def self.get_impl
+        ObjectSpace.each_object(Class).select { |klass| klass < self }
+      end
+    end
+
+    def create(name, uri)
+      log 'load journal storage adapter: ' + name
+      load File.join(File.dirname(__FILE__), 'journal', name + '.rb')
+      cls = AbstractStorage.get_impl.first
+      cls.new(uri: uri)
     end
   end
 
@@ -270,7 +286,12 @@ module Journal
       # overrides the internal consumer with a block
       @consumer = opts[:consumer]
       # storage backend
-      @storage = opts[:storage]
+      if @bot
+        @storage = opts[:storage] || Storage.create(
+            @bot.config['journal.storage'], @bot.config['journal.storage.uri'])
+      else
+        @storage = opts[:storage]
+      end
       @queue = Queue.new
       # consumer thread:
       @thread = Thread.new do
@@ -334,6 +355,17 @@ module Journal
       @subscriptions.delete subscription
     end
 
+    def find(query, limit=100, offset=0, &block)
+      if block_given?
+        begin
+          res = @storage.find(query, limit, offset)
+          block.call(res)
+        end until res.length > 0
+      else
+        @storage.find(query, limit, offset)
+      end
+    end
+
   end
 
 end # Journal
diff --git a/lib/rbot/journal/postgres.rb b/lib/rbot/journal/postgres.rb
index 57c539a7..65c67eb9 100644
--- a/lib/rbot/journal/postgres.rb
+++ b/lib/rbot/journal/postgres.rb
@@ -53,9 +53,14 @@ module Journal
           [m.id, m.topic, m.timestamp, JSON.generate(m.payload)])
       end
 
-      def find(query, limit=100, offset=0)
-        sql, params = query_to_sql(query)
-        sql = 'SELECT * FROM journal WHERE ' + sql + ' LIMIT %d OFFSET %d' % [limit.to_i, offset.to_i]
+      def find(query=nil, limit=100, offset=0)
+        if query
+          sql, params = query_to_sql(query)
+          sql = 'SELECT * FROM journal WHERE ' + sql + ' LIMIT %d OFFSET %d' % [limit.to_i, offset.to_i]
+        else
+          sql = 'SELECT * FROM journal LIMIT %d OFFSET %d' % [limit.to_i, offset.to_i]
+          params = []
+        end
         res = @conn.exec_params(sql, params)
         res.map do |row|
           timestamp = DateTime.strptime(row['timestamp'], '%Y-%m-%d %H:%M:%S%z')
@@ -65,13 +70,29 @@ module Journal
       end
 
       # returns the number of messages that match the query
-      def count(query)
-        sql, params = query_to_sql(query)
-        sql = 'SELECT COUNT(*) FROM journal WHERE ' + sql
+      def count(query=nil)
+        if query
+          sql, params = query_to_sql(query)
+          sql = 'SELECT COUNT(*) FROM journal WHERE ' + sql
+        else
+          sql = 'SELECT COUNT(*) FROM journal'
+          params = []
+        end
         res = @conn.exec_params(sql, params)
         res[0]['count'].to_i
       end
 
+      def remove(query=nil)
+        if query
+          sql, params = query_to_sql(query)
+          sql = 'DELETE FROM journal WHERE ' + sql
+        else
+          sql = 'DELETE FROM journal;'
+          params = []
+        end
+        res = @conn.exec_params(sql, params)
+      end
+
       def drop
         @conn.exec('DROP TABLE journal;') rescue nil
       end