Merge lp:~ajayaa/drizzle/event-notify-interface into lp:drizzle

[Ajaya Agrawal]

Added event_notify interface. Currently three types of event i.e. INSERT, UPDATE, DELETE in any type of table is observed. Client can issue wait_for() call to wait for certain number of events and then return.

[lp:~ajayaa/drizzle/event-notify-interface]

2649. By Ajaya Agrawal on 2013-09-18

changed slave code

[Stewart Smith (stewart)]

review: Approve

Unmerged revisions

2649. By Ajaya Agrawal on 2013-09-18

changed slave code

2648. By Ajaya Agrawal on 2013-09-12

added test cases for all type of events including insert in sys_replication_log and timeout

2647. By Ajaya Agrawal on 2013-09-12

added test cases for all type of events including insert in sys_replication_log and timeout

2646. By Ajaya Agrawal on 2013-09-11

fixed memory leak.working correctly. writing more test cases

2645. By Ajaya Agrawal on 2013-09-11

memory leak need to be fixed

2644. By Ajaya Agrawal on 2013-08-03

added a new event for replicationlog

2643. By Ajaya Agrawal on 2013-07-28

Added time_out support and test cases

2642. By Ajaya Agrawal on 2013-07-28

Added time_out support and test cases

2641. By Ajaya Agrawal on 2013-07-23

removed all hard coding and print statements and added destructor for event_t class

2640. By Ajaya Agrawal on 2013-07-23

removed all hard coding\n added destructor for event_t class

=== modified file 'drizzled/plugin/event_observer.cc'
--- drizzled/plugin/event_observer.cc	2011-03-22 12:01:19 +0000
+++ drizzled/plugin/event_observer.cc	2013-09-18 20:41:29 +0000
@@ -421,6 +421,13 @@
     return (in_table.getMutableShare()->getTableObservers() != NULL);
   }
 
+  // bool InsertToSysReplicationLogEventData::callEventObservers()
+  // {
+  //     EventObserverList *observers = new EventObserverList();
+  //     observers->addObserver(this, AFTER_INSERT_TO_SYS_REPLICATION_LOG);
+  
+  // }
+
   /*==========================================================*/
   /* Static meathods called by drizzle to notify interested plugins 
    * of a schema event.
@@ -599,6 +606,15 @@
     return eventData.callEventObservers();
   }
 
+  bool EventObserver::afterInsertToSysReplicationLog(Session &session)
+  {
+    if (all_event_plugins.empty())
+      return false;
+    InsertToSysReplicationLogEventData eventData(session);
+    return eventData.callEventObservers();
+   // std::cout << "event happening on sys_replication_log\n";
+    //return true;
+  }
 
 } /* namespace plugin */
 } /* namespace drizzled */

=== modified file 'drizzled/plugin/event_observer.h'
--- drizzled/plugin/event_observer.h	2011-08-14 17:04:01 +0000
+++ drizzled/plugin/event_observer.h	2013-09-18 20:41:29 +0000
@@ -79,6 +79,8 @@
     BEFORE_UPDATE_RECORD,   AFTER_UPDATE_RECORD, 
     BEFORE_DELETE_RECORD,   AFTER_DELETE_RECORD,
 
+    AFTER_INSERT_TO_SYS_REPLICATION_LOG,
+
     /* The max event ID marker. */
     MAX_EVENT_COUNT
   };
@@ -87,6 +89,10 @@
   {
     switch(event) 
     {
+
+    case AFTER_INSERT_TO_SYS_REPLICATION_LOG:
+      return "AFTER_INSERT_TO_SYS_REPLICATION_LOG";
+
     case BEFORE_DROP_TABLE:
       return "BEFORE_DROP_TABLE";
 
@@ -236,6 +242,8 @@
   static bool beforeDropDatabase(Session &session, const std::string &db);
   static bool afterDropDatabase(Session &session, const std::string &db, int err);
 
+  static bool afterInsertToSysReplicationLog(Session &session);
+
   static const EventObserverVector &getEventObservers(void);
 
 };
@@ -553,6 +561,15 @@
   {}  
 };
 
+class InsertToSysReplicationLogEventData: public SessionEventData
+{
+public:
+  InsertToSysReplicationLogEventData(Session &session_arg):
+    SessionEventData(EventObserver::AFTER_INSERT_TO_SYS_REPLICATION_LOG, session_arg)
+  {}
+ // virtual bool callEventObservers();  
+};
+
 //=======================================================
 
 } /* namespace plugin */

=== added directory 'plugin/event_notify'
=== added file 'plugin/event_notify/event_notify.cc'
--- plugin/event_notify/event_notify.cc	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/event_notify.cc	2013-09-18 20:41:29 +0000
@@ -0,0 +1,346 @@
+  /* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+   *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+   *
+   *  Copyright (C) Ajaya Agrawal
+   *
+   *  This program is free software; you can redistribute it and/or modify
+   *  it under the terms of the GNU General Public License as published by
+   *  the Free Software Foundation; version 2 of the License.
+   *
+   *  This program is distributed in the hope that it will be useful,
+   *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+   *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   *  GNU General Public License for more details.
+   *
+   *  You should have received a copy of the GNU General Public License
+   *  along with this program; if not, write to the Free Software
+   *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+   */
+
+  #include <config.h>
+  #include <fcntl.h>
+  #include <drizzled/function/str/strfunc.h>
+  #include <drizzled/item/func.h>
+  #include <drizzled/plugin.h>
+  #include <drizzled/plugin/function.h>
+  #include <drizzled/item.h>
+  #include <drizzled/module/option_map.h>
+  #include <drizzled/session.h>
+  #include <drizzled/session/times.h>
+  #include <drizzled/table/instance/base.h>
+  #include <drizzled/table.h>
+
+  #include "event_notify.h"
+
+  namespace po= boost::program_options;
+  using namespace std;
+  using namespace drizzled;
+  using namespace plugin;
+
+  static EventNotify *event_notify = NULL;
+
+  struct AddressIs {
+      event_t *ptr;
+      AddressIs(event_t *ptr) : ptr(ptr) {}
+      bool operator()(const event_t *object) const {
+          return ptr == object;
+      }
+  };
+
+  // void show()
+  // {
+  //   for (std::map<std::string, list<event_t*> >::iterator map_it=event_notify->all_events.begin(); map_it!=event_notify->all_events.end(); ++map_it){
+  //       std::cout << map_it->first << " "; 
+  //       for (list<event_t*>::iterator list_it=map_it->second.begin(); list_it!=map_it->second.end(); ++list_it)
+  //         std::cout << (*list_it)->event_type << " " << (*list_it)->time_out << " " << (*list_it)->num_of_events_threshold<< " " << (*list_it)->num_of_events_so_far << '\n';
+  //     }
+  // }
+
+  event_t::event_t(String *_table_name, String *_event_type, String *_time_out, String *_num_of_events_threshold)
+  {
+    this->table_name = _table_name->c_str();
+    this->event_type = _event_type->c_str();
+    time_out = atof(_time_out->c_str());
+    num_of_events_threshold = atoi(_num_of_events_threshold->c_str());
+    pthread_mutex_init(&this->lock, NULL);
+    pthread_cond_init(&this->cond, NULL);
+    num_of_events_so_far = 0;
+  }
+
+  event_t::~event_t()
+  {
+    pthread_mutex_destroy(&lock);
+    pthread_cond_destroy(&cond);
+  }
+
+  EventNotify::EventNotify(bool enabled):
+    drizzled::plugin::EventObserver("event_notify_interface"),
+    sysvar_enabled(enabled)
+  {
+  }
+
+  EventNotify::~EventNotify()
+  {
+  }
+
+  void EventNotify::registerTableEventsDo(TableShare &table_share, EventObserverList &observers)
+  {
+    boost::mutex::scoped_lock scoped(all_events_mutex);
+
+    string str(table_share.getTableName()); 
+    //cout << "in registertableeventsdo table name is " << str <<endl;
+    //show();
+    if (sysvar_enabled == false || (all_events.find(str) == all_events.end())){
+      return;
+    }
+    //cout << "yes event registered on this table\n";
+    registerEvent(observers, AFTER_INSERT_RECORD);
+    registerEvent(observers, AFTER_UPDATE_RECORD);
+    registerEvent(observers, AFTER_DELETE_RECORD);
+  }
+
+  void EventNotify::registerSessionEventsDo(Session &, EventObserverList &observers)
+  {
+    //here need to check whether drizzled was started with option --innodb.replication-log.
+    //cout << "registered for sys_replication_log\n";
+    registerEvent(observers, AFTER_INSERT_TO_SYS_REPLICATION_LOG);  
+  }
+
+  bool EventNotify::observeEventDo(EventData &data)
+  {
+    if (not sysvar_enabled)
+      return false;
+
+    switch (data.event) {
+
+    case AFTER_INSERT_RECORD:
+      afterInsertRecord((AfterInsertRecordEventData &)data);
+      break;
+
+    case AFTER_UPDATE_RECORD:
+      afterUpdateRecord((AfterUpdateRecordEventData &)data);
+      break;
+
+    case AFTER_DELETE_RECORD:
+      afterDeleteRecord((AfterDeleteRecordEventData &)data);
+      break;
+
+    case AFTER_INSERT_TO_SYS_REPLICATION_LOG:
+      afterInsertToSysReplicationLog((InsertToSysReplicationLogEventData&) data);
+      break;       
+
+    default:
+      fprintf(stderr, "event_notify: Unexpected event '%s'\n",
+        EventObserver::eventName(data.event));
+    }
+
+    return false;
+  }
+
+  bool EventNotify::afterInsertRecord(AfterInsertRecordEventData &data)
+  {
+    std::string str(data.table.getShare()->getTableName()); 
+    std::map<std::string, list<event_t*> >::iterator map_it;  
+    boost::mutex::scoped_lock scoped(all_events_mutex);
+    map_it = all_events.find(str);
+
+    if(map_it != all_events.end()){
+      for (list<event_t*>::iterator list_it=map_it->second.begin(); list_it!=map_it->second.end(); ++list_it){
+        if((*list_it)->event_type.compare("INSERT") == 0){
+          pthread_mutex_lock(&(*list_it)->lock);
+          (*list_it)->num_of_events_so_far++;
+          if((*list_it)->num_of_events_so_far >= (*list_it)->num_of_events_threshold){
+          pthread_cond_broadcast(&(*list_it)->cond);
+          fprintf(stderr, "broadcast %p\n", (void*)&(*list_it)->cond);
+        }
+          pthread_mutex_unlock(&(*list_it)->lock);
+        }
+      }
+    }
+    return false; 
+  }
+
+  bool EventNotify::afterUpdateRecord(AfterUpdateRecordEventData &data)
+  {
+    std::string str(data.table.getShare()->getTableName()); 
+    std::map<std::string, list<event_t*> >::iterator map_it;  
+    boost::mutex::scoped_lock scoped(all_events_mutex);
+    map_it = all_events.find(str);
+
+    if(map_it != all_events.end()){
+      for (list<event_t*>::iterator list_it=map_it->second.begin(); list_it!=map_it->second.end(); ++list_it){
+        if((*list_it)->event_type.compare("UPDATE") == 0){
+          pthread_mutex_lock(&(*list_it)->lock);
+          (*list_it)->num_of_events_so_far++;
+          if((*list_it)->num_of_events_so_far >= (*list_it)->num_of_events_threshold){
+          pthread_cond_broadcast(&(*list_it)->cond);
+          fprintf(stderr, "broadcast %p\n", (void*)&(*list_it)->cond);
+        }
+          pthread_mutex_unlock(&(*list_it)->lock);
+        }
+      }
+    }
+    return false; 
+  }
+
+  bool EventNotify::afterDeleteRecord(AfterDeleteRecordEventData &data)
+  {
+    std::string str(data.table.getShare()->getTableName()); 
+    std::map<std::string, list<event_t*> >::iterator map_it;  
+    boost::mutex::scoped_lock scoped(all_events_mutex);
+    map_it = all_events.find(str);
+
+    if(map_it != all_events.end()){
+      for (list<event_t*>::iterator list_it=map_it->second.begin(); list_it!=map_it->second.end(); ++list_it){
+        if((*list_it)->event_type.compare("DELETE") == 0){
+          pthread_mutex_lock(&(*list_it)->lock);
+          (*list_it)->num_of_events_so_far++;
+          if((*list_it)->num_of_events_so_far >= (*list_it)->num_of_events_threshold){
+          pthread_cond_broadcast(&(*list_it)->cond);
+          fprintf(stderr, "broadcast %p\n", (void*)&(*list_it)->cond);
+        }
+          pthread_mutex_unlock(&(*list_it)->lock);
+        }
+      }
+    }
+    return false; 
+  }
+
+
+  bool EventNotify::afterInsertToSysReplicationLog(InsertToSysReplicationLogEventData &data)
+  {
+    std::string str = "sys_replication_log";
+    std::map<std::string, list<event_t*> >::iterator map_it;  
+    boost::mutex::scoped_lock scoped(all_events_mutex);
+    map_it = all_events.find(str);
+
+    if(map_it != all_events.end()){
+      for (list<event_t*>::iterator list_it=map_it->second.begin(); list_it!=map_it->second.end(); ++list_it){
+        if((*list_it)->event_type.compare("INSERT") == 0){
+          pthread_mutex_lock(&(*list_it)->lock);
+          (*list_it)->num_of_events_so_far++;
+          if((*list_it)->num_of_events_so_far >= (*list_it)->num_of_events_threshold){
+            pthread_cond_broadcast(&(*list_it)->cond);
+            fprintf(stderr, "broadcast %p\n", (void*)&(*list_it)->cond);
+          }
+          pthread_mutex_unlock(&(*list_it)->lock);
+        }
+      }
+    }
+    return false;
+  }
+
+
+  class Item_func_event_notify : public Item_str_func
+  {
+  public:
+    Item_func_event_notify() : Item_str_func() {}
+    const char *func_name() const { return "wait_for"; }
+
+    String *val_str(String* s) 
+    {
+      String *table_name = args[0]->val_str(s);
+      event_t *new_event = new event_t(args[0]->val_str(s),(args[1]->val_str(s)), (args[2]->val_str(s)), (args[3]->val_str(s)));
+
+      map<std::string, list<event_t*> >::iterator map_it;
+      list<event_t*>::iterator list_it;
+
+      {
+        boost::mutex::scoped_lock scoped(event_notify->all_events_mutex);
+        map_it = event_notify->all_events.find(new_event->table_name);
+
+        if(map_it == event_notify->all_events.end()){
+          list<event_t*> my_list;
+          my_list.push_back(new_event);
+          event_notify->all_events[new_event->table_name] = my_list;
+        }
+        else{
+           for (list_it=map_it->second.begin(); list_it!=map_it->second.end(); ++list_it){
+            if((*list_it)->event_type.compare(new_event->event_type) == 0 ) {
+                if( new_event->num_of_events_threshold <= (*list_it)->num_of_events_so_far){
+                  //int retval = new_event->num_of_events_threshold;
+                  delete new_event;
+                  return table_name;
+                }
+                new_event->num_of_events_so_far = (*list_it)->num_of_events_so_far;
+                break;
+            }
+          }
+         map_it->second.push_back(new_event);
+        }
+      } 
+  /*
+  * waiting for number of events reach its threshold.
+  */  
+      struct timespec timeToWait;
+      struct timeval now;
+      gettimeofday(&now,NULL);
+      timeToWait.tv_sec = now.tv_sec+new_event->time_out;
+      pthread_mutex_lock(&new_event->lock);
+      //while(new_event->num_of_events_so_far < new_event->num_of_events_threshold )
+      fprintf(stderr, "wait %p\n", (void*)&new_event->cond);
+      pthread_cond_timedwait(&new_event->cond, &new_event->lock, &timeToWait);
+      fprintf(stderr, "SIGNALLED %p\n", (void*)&new_event->cond);
+
+      pthread_mutex_unlock(&new_event->lock);
+      int count_events = 0;
+        {
+        boost::mutex::scoped_lock scoped(event_notify->all_events_mutex);
+        map<std::string, list<event_t*> >::iterator map_iter;
+        list<event_t*>::iterator list_iter;
+        map_iter = event_notify->all_events.find(new_event->table_name);
+        for (list_iter=map_iter->second.begin(); list_iter!=map_iter->second.end(); ++list_iter){
+            if((*list_iter)->event_type.compare(new_event->event_type) == 0 ){
+              count_events++;
+            }
+        }
+        if(count_events > 1){
+          map_iter->second.remove_if(AddressIs(new_event));
+          delete new_event;
+        }
+      }
+      return table_name;
+    }
+
+    void fix_length_and_dec() {
+      max_length=strlen("wait_for");
+    }
+
+    bool check_argument_count(int n) 
+    { 
+      return (n == 4); 
+    }
+  };
+
+  plugin::Create_function<Item_func_event_notify> *event_notify_udf= NULL;
+
+  static void init_options(drizzled::module::option_context &)
+  {
+    event_notify = new EventNotify(true);
+  }
+
+  static int event_notifier_plugin_init(drizzled::module::Context &context)
+  {
+
+    event_notify_udf=
+      new plugin::Create_function<Item_func_event_notify>("wait_for");
+    context.add(event_notify_udf);
+    context.add(event_notify);
+    context.registerVariable(new sys_var_bool_ptr("enabled", &event_notify->sysvar_enabled));
+    return 0;
+  }
+
+
+  DRIZZLE_DECLARE_PLUGIN
+  {
+    DRIZZLE_VERSION_ID,            /* DRIZZLE_VERSION_ID */
+    "Event_Notifier",                 /* module name */
+    "1.0",                         /* module version */
+    "Ajaya Agrawal",                 /* author(s) */
+    N_("indicates an event happening on certain table/database/session"), /* description */
+    PLUGIN_LICENSE_BSD,            /* license */
+    event_notifier_plugin_init,       /* init module function */
+    NULL,                          /* module dependencies */
+    init_options                           /* init options function */
+  }
+  DRIZZLE_DECLARE_PLUGIN_END;
\ No newline at end of file

=== added file 'plugin/event_notify/event_notify.h'
--- plugin/event_notify/event_notify.h	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/event_notify.h	2013-09-18 20:41:29 +0000
@@ -0,0 +1,81 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright 2013 Ajaya Agrawal
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <drizzled/plugin/event_observer.h>
+#include <boost/thread/mutex.hpp>
+
+#include <map>
+#include <list>
+#include <pthread.h>
+#include <string>
+#include <cstring>
+
+namespace drizzled {
+namespace plugin {
+ 
+struct cmp_str
+{
+ bool operator()(std::string a, std::string b)
+ {
+    return (a.compare(b) < 0);
+ }
+};
+
+class event_t {
+
+public:
+  event_t(String *_session_time, String *_event_type, String *_table_name, String *_num_of_events);
+  ~event_t();
+   std::string table_name;
+   std::string event_type;  
+   pthread_mutex_t lock;
+   pthread_cond_t cond;
+   int num_of_events_threshold;
+   double time_out; 
+   int num_of_events_so_far; 
+};
+
+class EventNotify : public drizzled::plugin::EventObserver
+{
+public:
+  EventNotify(bool enabled);
+  ~EventNotify();
+  
+  void registerTableEventsDo(TableShare &table_share, EventObserverList &observers);
+  void registerSessionEventsDo(Session &session, EventObserverList &observers);
+  bool observeEventDo(EventData &);
+  bool afterStatement(AfterStatementEventData &data);
+  bool afterInsertRecord(AfterInsertRecordEventData &data); 
+  bool afterUpdateRecord(AfterUpdateRecordEventData &data);
+  bool afterDeleteRecord(AfterDeleteRecordEventData &data);
+  bool afterInsertToSysReplicationLog(InsertToSysReplicationLogEventData &data);
+  /**
+   * These are the event_notify system variables.  So sysvar_enabled is
+   * event_notify_enabled in SHOW VARIABLES, etc.  They are all global and dynamic.
+   */
+  bool sysvar_enabled;
+  boost::mutex all_events_mutex;
+  boost::mutex counter_mutex;
+  std::map <std::string, std::list<event_t*>, cmp_str > all_events;
+};
+
+} /* namespace plugin */
+} /* namespace drizzled */

=== added file 'plugin/event_notify/plugin.ini'
--- plugin/event_notify/plugin.ini	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/plugin.ini	2013-09-18 20:41:29 +0000
@@ -0,0 +1,3 @@
+[plugin]
+sources=event_notify.cc
+headers=event_notify.h

=== added directory 'plugin/event_notify/tests'
=== added directory 'plugin/event_notify/tests/r'
=== added file 'plugin/event_notify/tests/r/delete.result'
--- plugin/event_notify/tests/r/delete.result	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/r/delete.result	2013-09-18 20:41:29 +0000
@@ -0,0 +1,13 @@
+drop table if exists t1;
+create table t1 (a int primary key);
+select wait_for("t1", "DELETE", "200.9", "2");
+insert into t1 values(1),(2),(3);
+delete from t1 where a=1;
+delete from t1 where a=2;
+delete from t1 where a=3;
+wait_for("t1", "DELETE", "200.9", "2")
+t1
+select wait_for("t1", "DELETE", "1000", "3");
+wait_for("t1", "DELETE", "1000", "3")
+t1
+drop table t1;

=== added file 'plugin/event_notify/tests/r/insert.result'
--- plugin/event_notify/tests/r/insert.result	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/r/insert.result	2013-09-18 20:41:29 +0000
@@ -0,0 +1,18 @@
+drop table if exists t1;
+create table t1 (a int primary key);
+select wait_for("t1", "INSERT", "100", "4");
+insert into t1 values(1);
+insert into t1 values(2);
+insert into t1 values(3);
+insert into t1 values(4);
+wait_for("t1", "INSERT", "100", "4")
+t1
+select wait_for("t1", "INSERT", "100", "5");
+insert into t1 values(5);
+wait_for("t1", "INSERT", "100", "5")
+t1
+insert into t1 values(60);
+select wait_for("t1", "INSERT", "100", "6");
+wait_for("t1", "INSERT", "100", "6")
+t1
+drop table t1;

=== added file 'plugin/event_notify/tests/r/sys_replication_log.result'
--- plugin/event_notify/tests/r/sys_replication_log.result	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/r/sys_replication_log.result	2013-09-18 20:41:29 +0000
@@ -0,0 +1,12 @@
+drop table if exists t1;
+create table t1 (a int primary key);
+select wait_for("sys_replication_log", "INSERT", "200.9", "1");
+insert into t1 values(1);
+insert into t1 values(2);
+insert into t1 values(3);
+wait_for("sys_replication_log", "INSERT", "200.9", "1")
+sys_replication_log
+select wait_for("sys_replication_log", "INSERT", "100", "3");
+wait_for("sys_replication_log", "INSERT", "100", "3")
+sys_replication_log
+drop table t1;

=== added file 'plugin/event_notify/tests/r/time_out.result'
--- plugin/event_notify/tests/r/time_out.result	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/r/time_out.result	2013-09-18 20:41:29 +0000
@@ -0,0 +1,3 @@
+select wait_for("t1", "INSERT", "5", "20");
+wait_for("t1", "INSERT", "5", "20")
+t1

=== added file 'plugin/event_notify/tests/r/update.result'
--- plugin/event_notify/tests/r/update.result	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/r/update.result	2013-09-18 20:41:29 +0000
@@ -0,0 +1,13 @@
+drop table if exists t1;
+create table t1 (a int primary key);
+select wait_for("t1", "UPDATE", "200.9", "1");
+insert into t1 values(1),(2),(3);
+update t1 set a=10 where a=1;
+update t1 set a=20 where a=2;
+update t1 set a=30 where a=3;
+wait_for("t1", "UPDATE", "200.9", "1")
+t1
+select wait_for("t1", "UPDATE", "1000", "2");
+wait_for("t1", "UPDATE", "1000", "2")
+t1
+drop table t1;

=== added directory 'plugin/event_notify/tests/t'
=== added file 'plugin/event_notify/tests/t/delete.test'
--- plugin/event_notify/tests/t/delete.test	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/t/delete.test	2013-09-18 20:41:29 +0000
@@ -0,0 +1,21 @@
+connect (con1,localhost,root,,);
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+create table t1 (a int primary key);
+connection con1;
+send select wait_for("t1", "DELETE", "200.9", "2");
+connection default;
+insert into t1 values(1),(2),(3);
+delete from t1 where a=1;
+delete from t1 where a=2;
+delete from t1 where a=3;
+connection con1;
+reap;
+send select wait_for("t1", "DELETE", "1000", "3");
+connection default;
+connection con1;
+reap;
+disconnect con1;
+connection default;
+drop table t1;

=== added file 'plugin/event_notify/tests/t/insert.test'
--- plugin/event_notify/tests/t/insert.test	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/t/insert.test	2013-09-18 20:41:29 +0000
@@ -0,0 +1,28 @@
+connect (con1,localhost,root,,);
+connect (con2,localhost,root,,);
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+create table t1 (a int primary key);
+connection con1;
+send select wait_for("t1", "INSERT", "100", "4");
+connection default;
+insert into t1 values(1);
+insert into t1 values(2);
+insert into t1 values(3);
+insert into t1 values(4);
+connection con1;
+reap;
+connection con2;
+send select wait_for("t1", "INSERT", "100", "5");
+connection default;
+insert into t1 values(5);
+connection con2;
+reap;
+insert into t1 values(60);
+send select wait_for("t1", "INSERT", "100", "6");
+reap;
+disconnect con1;
+disconnect con2;
+connection default;
+drop table t1;

=== added file 'plugin/event_notify/tests/t/master.opt'
--- plugin/event_notify/tests/t/master.opt	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/t/master.opt	2013-09-18 20:41:29 +0000
@@ -0,0 +1,1 @@
+--plugin-add=event_notify --innodb.replication-log

=== added file 'plugin/event_notify/tests/t/sys_replication_log.test'
--- plugin/event_notify/tests/t/sys_replication_log.test	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/t/sys_replication_log.test	2013-09-18 20:41:29 +0000
@@ -0,0 +1,20 @@
+connect (con1,localhost,root,,);
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+create table t1 (a int primary key);
+connection con1;
+send select wait_for("sys_replication_log", "INSERT", "200.9", "1");
+connection default;
+insert into t1 values(1);
+insert into t1 values(2);
+insert into t1 values(3);
+connection con1;
+reap;
+send select wait_for("sys_replication_log", "INSERT", "100", "3");
+connection default;
+connection con1;
+reap;
+disconnect con1;
+connection default;
+drop table t1;

=== added file 'plugin/event_notify/tests/t/time_out.test'
--- plugin/event_notify/tests/t/time_out.test	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/t/time_out.test	2013-09-18 20:41:29 +0000
@@ -0,0 +1,2 @@
+send select wait_for("t1", "INSERT", "5", "20");
+reap;

=== added file 'plugin/event_notify/tests/t/update.test'
--- plugin/event_notify/tests/t/update.test	1970-01-01 00:00:00 +0000
+++ plugin/event_notify/tests/t/update.test	2013-09-18 20:41:29 +0000
@@ -0,0 +1,21 @@
+connect (con1,localhost,root,,);
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+create table t1 (a int primary key);
+connection con1;
+send select wait_for("t1", "UPDATE", "200.9", "1");
+connection default;
+insert into t1 values(1),(2),(3);
+update t1 set a=10 where a=1;
+update t1 set a=20 where a=2;
+update t1 set a=30 where a=3;
+connection con1;
+reap;
+send select wait_for("t1", "UPDATE", "1000", "2");
+connection default;
+connection con1;
+reap;
+disconnect con1;
+connection default;
+drop table t1;

=== modified file 'plugin/slave/queue_producer.cc'
--- plugin/slave/queue_producer.cc	2012-04-14 20:43:20 +0000
+++ plugin/slave/queue_producer.cc	2013-09-18 20:41:29 +0000
@@ -49,6 +49,62 @@
   return reconnect(true);
 }
 
+void QueueProducer::run()
+{
+  boost::posix_time::seconds duration(getSleepInterval());
+
+  /* thread setup needed to do things like create a Session */
+  internal::my_thread_init();
+
+  if (not init())
+    return;
+
+  while (1)
+  {
+    {
+      /* This uninterruptable block processes the message queue */
+      boost::this_thread::disable_interruption di;
+
+      if (not process())
+      {
+        shutdown();
+        return;
+      }
+    }
+
+    /* Interruptable only when not doing work (aka, sleeping) */
+    try
+    {
+      /* waiting for num_of_events through a wait_for() call*/
+      int num_of_events = 2;
+      std::string event_sql("select wait_for(\"sys_replication_log\", \"INSERT\", \"2000\", \"");
+      event_sql.append(boost::lexical_cast<string>(num_of_events));
+      event_sql.append("\")");
+      drizzle_return_t event_ret;
+      drizzle_result_st event_result;
+      drizzle_query_str(_connection, &event_result, event_sql.c_str(), &event_ret);
+  
+      if (event_ret != DRIZZLE_RETURN_OK){
+        drizzle_result_free(&event_result);
+        boost::this_thread::sleep(duration);
+        string error = "";
+        error.append(drizzle_error(_drizzle));        
+        cout << error <<": producer thread sleeping\n";        
+      }
+      else{
+        num_of_events += 1;
+        cout << "wait_for returned\n";
+        drizzle_result_free(&event_result);
+      }
+    }
+    catch (boost::thread_interrupted &)
+    {
+      return;
+    }
+  }
+}
+
+
 bool QueueProducer::process()
 {
   if (_saved_max_commit_id == 0)

=== modified file 'plugin/slave/queue_thread.h'
--- plugin/slave/queue_thread.h	2011-03-14 05:40:28 +0000
+++ plugin/slave/queue_thread.h	2013-09-18 20:41:29 +0000
@@ -40,7 +40,7 @@
   virtual ~QueueThread()
   {}
   
-  void run(void);
+  virtual void run(void);
 
   /**
    * Do any initialization work.