From 6ee094af95747f65800c9e822e5566f1c5cf0bfe Mon Sep 17 00:00:00 2001
From: Dominik Sander <git@dsander.de>
Date: Sun, 7 Feb 2016 12:38:36 +0100
Subject: [PATCH] TwitterStreamAgent prevent connection in use and restart
 issues

Ensures the database connection is checked back in to the pool before restarting.
Restart TwitterStreamAgent outside of the Rufus::Scheduler work thread.
---
 app/concerns/long_runnable.rb             |  4 +++-
 app/models/agents/twitter_stream_agent.rb | 11 ++++++-----
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/app/concerns/long_runnable.rb b/app/concerns/long_runnable.rb
index 3f01be8cf6..d217d01c5b 100644
--- a/app/concerns/long_runnable.rb
+++ b/app/concerns/long_runnable.rb
@@ -98,13 +98,15 @@ def stop!
 
     def terminate_thread!
       if thread
-        thread.terminate
+        thread.instance_eval { ActiveRecord::Base.connection_pool.release_connection }
         thread.wakeup if thread.status == 'sleep'
+        thread.terminate
       end
     end
 
     def restart!
       without_alive_check do
+        puts "--> Restarting #{id} at #{Time.now} <--"
         stop!
         setup!(scheduler, mutex)
         run!
diff --git a/app/models/agents/twitter_stream_agent.rb b/app/models/agents/twitter_stream_agent.rb
index a51d716a79..0822881328 100644
--- a/app/models/agents/twitter_stream_agent.rb
+++ b/app/models/agents/twitter_stream_agent.rb
@@ -157,16 +157,14 @@ class Worker < LongRunnable::Worker
       def setup
         require 'twitter/json_stream'
         @filter_to_agent_map = @config[:filter_to_agent_map]
-
-        schedule_in RELOAD_TIMEOUT do
-          puts "--> Restarting TwitterStream #{id} at #{Time.now} <--"
-          restart!
-        end
       end
 
       def run
         @recent_tweets = []
         EventMachine.run do
+          EventMachine.add_periodic_timer(RELOAD_TIMEOUT) do
+            restart!
+          end
           stream!(@filter_to_agent_map.keys, @agent) do |status|
             handle_status(status)
           end
@@ -200,6 +198,9 @@ def stream!(filters, agent, &block)
 
         stream.on_error do |message|
           STDERR.puts " --> Twitter error: #{message} at #{Time.now} <--"
+          STDERR.puts " --> Sleeping for 15 seconds"
+          sleep 15
+          restart!
         end
 
         stream.on_no_data do |message|