Merge lp:~jaypipes/drizzle/publisher into lp:~drizzle-trunk/drizzle/development

[Jay Pipes]

First patchset adding basic publisher support to the
replication services component.

A Publisher tracks what transactions have been applied to a channel on
a node and provides an interface for querying for this
information. Later, the publisher will act as a specialized
server which connects to a Subscriber plugin.

Adds a new replication_publishers data_dictionary view,
mostly used in testing to see if publisher registration
was correct.

Adds a new replication_publisher_channels data_dictionary view, again
mostly used in testing registration.

[Jay Pipes (jaypipes)]

Taking into Work In Progress. Eric and I discussed moving the replication.proto into the transaction log module and not requiring publishers and subscribers to keep a GPB manifest if they didn't want to. This makes sense. The GPB manifest is implementation, not interface.

[Jay Pipes (jaypipes)]

Moves the replication.proto file into /plugin/transaction_log/ since it is implementation-dependent stuff.

[lp:~jaypipes/drizzle/publisher]

1440. By Jay Pipes <jpipes@serialcoder> on 2010-04-26

Publisher manifest file now written to disk properly (was using incorrect ByteSize() instead of std::string::size(). Also, manifest is read on startup

1441. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Adds a custom test suite and framework for testing the replication pieces. The first test case tests basic functionality of creating and writing a serialized publisher manifest file.

1442. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Connect replication test suite into main transaction_log test suite

1443. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Adds test cases to replication test suite for checking
startup when a corrupted manifest file is present, as
well as a good publisher manifest is present. Changes
the TransactionLogPublisher constructor to throw a proper
error and disable the publisher plugin properly.

[Eric Day (eday)]

Great work! Plugin point for Publisher looks sane, as for locking in the transaction log publisher plugin (your comment on using a scoreboard method), it depends how expensive the calls are in the lock. We might just want to break it down to one lock/channel.

[Eric Day (eday)]

review: Approve

[lp:~jaypipes/drizzle/publisher]

1444. By Jay Pipes <jpipes@serialcoder> on 2010-05-01

Fix valgrind issue due to not initializing the publisher's mutex before calling readManifestFile()

[Jay Pipes (jaypipes)]

FYI, saw the valgrind error and have pushed a fix.

[lp:~jaypipes/drizzle/publisher]

1445. By Jay Pipes <jpipes@serialcoder> on 2010-05-04

Merge trunk

[Brian Aker (brianaker)]

review: Needs Information

[Jay Pipes (jaypipes)]

Heya, what information do you need on this one? Related to the
testing in Python or something else? Happy to answer, just let me
know...

On Thu, May 27, 2010 at 1:07 AM, Brian Aker <email address hidden> wrote:
> Review: Needs Information
>
> --
> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
> You are the owner of lp:~jaypipes/drizzle/publisher.
>

[Brian Aker (brianaker)]

Hi!

I marked it because of the python testing. PatrickC offered to look at extend test-run.pl so solve this issue so I am currently waiting to hear from him about it.

Cheers,
 -Brian

On May 27, 2010, at 8:42 AM, Jay Pipes wrote:

> Heya, what information do you need on this one? Related to the
> testing in Python or something else? Happy to answer, just let me
> know...
>
> On Thu, May 27, 2010 at 1:07 AM, Brian Aker <email address hidden> wrote:
>> Review: Needs Information
>>
>> --
>> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
>> You are the owner of lp:~jaypipes/drizzle/publisher.
>>
> --
> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
> You are reviewing the proposed merge of lp:~jaypipes/drizzle/publisher into lp:drizzle.

[Patrick Crews (patrick-crews)]

Expect to have some word on this next week. I've been a bit occupied with
some other tasks this week. I'll likely be in touch with you then, Jay. My
apologies for the hold-up on the review.

On Fri, May 28, 2010 at 6:34 PM, Brian Aker <email address hidden> wrote:

> Hi!
>
> I marked it because of the python testing. PatrickC offered to look at
> extend test-run.pl so solve this issue so I am currently waiting to hear
> from him about it.
>
> Cheers,
> -Brian
>
> On May 27, 2010, at 8:42 AM, Jay Pipes wrote:
>
> > Heya, what information do you need on this one? Related to the
> > testing in Python or something else? Happy to answer, just let me
> > know...
> >
> > On Thu, May 27, 2010 at 1:07 AM, Brian Aker <email address hidden> wrote:
> >> Review: Needs Information
> >>
> >> --
> >> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
> >> You are the owner of lp:~jaypipes/drizzle/publisher.
> >>
> > --
> > https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
> > You are reviewing the proposed merge of lp:~jaypipes/drizzle/publisher
> into lp:drizzle.
>
> --
> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
> Your team Drizzle-developers is requested to review the proposed merge of
> lp:~jaypipes/drizzle/publisher into lp:drizzle.
>

[Jay Pipes (jaypipes)]

No worries at all, Patrick.

On Fri, May 28, 2010 at 6:39 PM, Patrick Crews <email address hidden> wrote:
> Expect to have some word on this next week.  I've been a bit occupied with
> some other tasks this week.  I'll likely be in touch with you then, Jay.  My
> apologies for the hold-up on the review.
>
> On Fri, May 28, 2010 at 6:34 PM, Brian Aker <email address hidden> wrote:
>
>> Hi!
>>
>> I marked it because of the python testing. PatrickC offered to look at
>> extend test-run.pl so solve this issue so I am currently waiting to hear
>> from him about it.
>>
>> Cheers,
>>        -Brian
>>
>> On May 27, 2010, at 8:42 AM, Jay Pipes wrote:
>>
>> > Heya, what information do you need on this one?  Related to the
>> > testing in Python or something else?  Happy to answer, just let me
>> > know...
>> >
>> > On Thu, May 27, 2010 at 1:07 AM, Brian Aker <email address hidden> wrote:
>> >> Review: Needs Information
>> >>
>> >> --
>> >> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
>> >> You are the owner of lp:~jaypipes/drizzle/publisher.
>> >>
>> > --
>> > https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
>> > You are reviewing the proposed merge of lp:~jaypipes/drizzle/publisher
>> into lp:drizzle.
>>
>> --
>> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
>> Your team Drizzle-developers is requested to review the proposed merge of
>> lp:~jaypipes/drizzle/publisher into lp:drizzle.
>>
>
> --
> https://code.launchpad.net/~jaypipes/drizzle/publisher/+merge/23785
> You are the owner of lp:~jaypipes/drizzle/publisher.
>

[Stewart Smith (stewart)]

On Fri, 28 May 2010 22:39:30 -0000, Patrick Crews <email address hidden> wrote:
> Expect to have some word on this next week. I've been a bit occupied with
> some other tasks this week. I'll likely be in touch with you then, Jay. My
> apologies for the hold-up on the review.

We should really port mtr2 back in before doing lots of modifications.

--
Stewart Smith

[Patrick Crews (patrick-crews)]

After looking at the tests here, I have to say that I'm in favor of leaving the Python testing as-is unless there is some really urgent need to not have it.

My reasoning:
* protoc doesn't produce perl code. We use the python-generated code to interact with the manifest and get useful, test-related data.
* The current python code is awfully clean and assertion based, which can fill in gaps that test-run doesn't cover effectively. It's also well-developed / not-half-baked and we can expand the Python test without a lot of fuss.

If the use of python is a deal-breaker, we probably could do something to make it work, but I fear that the effort involved would not be worth the pay-off (at least at present).

I will be looking at - https://bugs.launchpad.net/drizzle/+bug/570948 - bad command-line options hang test cases so that we can improve test-run, but it's just my feeling that we can move things forward by keeping the current test(s)

[Stewart Smith (stewart)]

On Thu, 03 Jun 2010 16:03:26 -0000, Patrick Crews <email address hidden> wrote:
> If the use of python is a deal-breaker, we probably could do something to make it work, but I fear that the effort involved would not be worth the pay-off (at least at present).

We rely on python for parts of the build system anyway, so it's not a
show stopper (if it runs on the pre-historic versions that solaris ship with)

--
Stewart Smith

[Jay Pipes (jaypipes)]

Not sure why we even are discussing this. Brian's made a decision and that's apparently all that matters.

[Brian Aker (brianaker)]

Hi!

Until it fits under the test framework we have it won't be mergable. As pointed out in previous comments "and we can expand the Python test without a lot of fuss" is exactly the issue. Once we have two systems we end up with a tower babble. So until we can test with the current framework this patch is non-starter.

Cheers,
  -Brian

review: Disapprove

Unmerged revisions

1445. By Jay Pipes <jpipes@serialcoder> on 2010-05-04

Merge trunk

1444. By Jay Pipes <jpipes@serialcoder> on 2010-05-01

Fix valgrind issue due to not initializing the publisher's mutex before calling readManifestFile()

1443. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Adds test cases to replication test suite for checking
startup when a corrupted manifest file is present, as
well as a good publisher manifest is present. Changes
the TransactionLogPublisher constructor to throw a proper
error and disable the publisher plugin properly.

1442. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Connect replication test suite into main transaction_log test suite

1441. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Adds a custom test suite and framework for testing the replication pieces. The first test case tests basic functionality of creating and writing a serialized publisher manifest file.

1440. By Jay Pipes <jpipes@serialcoder> on 2010-04-26

Publisher manifest file now written to disk properly (was using incorrect ByteSize() instead of std::string::size(). Also, manifest is read on startup

1439. By Jay Pipes <jpipes@serialcoder> on 2010-04-26

Merge Monty build fixes

1438. By Jay Pipes <jpipes@serialcoder> on 2010-04-23

remove references to /drizzled/message/replication.pb.h

1437. By Jay Pipes <jpipes@serialcoder> on 2010-04-23

Moves replication.proto out of /drizzled/message and into /plugin/transaction_log/ since it is implementation-specific.

1436. By Jay Pipes <jpipes@serialcoder> on 2010-04-20

Remove unused ZeroCopyStream buffer

=== modified file '.bzrignore'
--- .bzrignore	2010-02-24 00:19:06 +0000
+++ .bzrignore	2010-05-05 02:25:42 +0000
@@ -304,3 +304,4 @@
 *.reject
 config/pandora_vc_revinfo
 drizzled/message/schema_writer
+replication_pb2.py

=== modified file 'drizzled/include.am'
--- drizzled/include.am	2010-04-27 21:03:13 +0000
+++ drizzled/include.am	2010-05-05 02:25:42 +0000
@@ -306,6 +306,7 @@
 			      drizzled/plugin/monitored_in_transaction.h \
 			      drizzled/plugin/null_client.h \
 			      drizzled/plugin/plugin.h \
+			      drizzled/plugin/publisher.h \
 			      drizzled/plugin/query_cache.h \
 			      drizzled/plugin/query_rewrite.h \
 			      drizzled/plugin/replication.h \
@@ -677,6 +678,7 @@
 				 drizzled/plugin/logging.cc \
 				 drizzled/plugin/monitored_in_transaction.cc \
 				 drizzled/plugin/plugin.cc \
+				 drizzled/plugin/publisher.cc \
 				 drizzled/plugin/query_cache.cc \
 				 drizzled/plugin/query_rewrite.cc \
 				 drizzled/plugin/registry.cc \

=== modified file 'drizzled/message/transaction.proto'
--- drizzled/message/transaction.proto	2010-02-26 00:05:00 +0000
+++ drizzled/message/transaction.proto	2010-05-05 02:25:42 +0000
@@ -304,9 +304,10 @@
 message TransactionContext
 {
   required uint32 server_id = 1; /* Unique identifier of a server */
-  required uint64 transaction_id = 2; /* Globally-unique transaction ID */
+  required uint64 transaction_id = 2; /* Unique transaction ID within the channel */
   required uint64 start_timestamp = 3; /* Timestamp of when the transaction started */
   required uint64 end_timestamp = 4; /* Timestamp of when the transaction ended */
+  optional uint32 channel_id = 5 [default = 1]; /* Scope in which transaction ID is unique */
 }
 
 /*

=== added file 'drizzled/plugin/publisher.cc'
--- drizzled/plugin/publisher.cc	1970-01-01 00:00:00 +0000
+++ drizzled/plugin/publisher.cc	2010-05-05 02:25:42 +0000
@@ -0,0 +1,47 @@
+/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (c) 2010 Jay Pipes
+ *
+ *  Authors:
+ *
+ *    Jay Pipes <[e-mail]>
+ *
+ *  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 "drizzled/plugin/publisher.h"
+#include "drizzled/replication_services.h"
+
+namespace drizzled
+{
+namespace plugin
+{
+
+bool plugin::Publisher::addPlugin(plugin::Publisher *publisher)
+{
+  ReplicationServices &replication_services= ReplicationServices::singleton();
+  replication_services.attachPublisher(publisher);
+  return false;
+}
+
+void plugin::Publisher::removePlugin(plugin::Publisher *publisher)
+{
+  ReplicationServices &replication_services= ReplicationServices::singleton();
+  replication_services.detachPublisher(publisher);
+}
+
+} // namespace plugin
+} // namespace drizzled

=== added file 'drizzled/plugin/publisher.h'
--- drizzled/plugin/publisher.h	1970-01-01 00:00:00 +0000
+++ drizzled/plugin/publisher.h	2010-05-05 02:25:42 +0000
@@ -0,0 +1,99 @@
+/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (c) 2010 Jay Pipes
+ *
+ *  Authors:
+ *
+ *    Jay Pipes <[e-mail]>
+ *
+ *  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
+ */
+
+#ifndef DRIZZLED_PLUGIN_PUBLISHER_H
+#define DRIZZLED_PLUGIN_PUBLISHER_H
+
+/**
+ * @file Defines the API for a Publisher
+ *
+ * A Publisher keeps track of transactions that have been applied
+ * to a data store and publishes information about the transactions to
+ * a Subscriber.
+ */
+
+#include "drizzled/plugin/plugin.h"
+
+#include "drizzled/transaction_services.h"
+#include "drizzled/message/transaction.pb.h"
+
+#include <functional>
+
+namespace drizzled
+{
+
+namespace plugin
+{
+
+/**
+ * A class responsible for publishing replication information
+ * to subscribers.
+ */
+class Publisher :public Plugin
+{
+public:
+  explicit Publisher(std::string name_arg) :
+    Plugin(name_arg, "Publisher")
+  {}
+  virtual ~Publisher() {}
+  /**
+   * Returns the transaction ID for the last message that was
+   * successfully applied on the publisher node for a given
+   * channel.
+   *
+   * @param[in] Channel ID to check
+   */
+  virtual TransactionServices::TransactionId getLastAppliedTransactionId(uint32_t channel_id)= 0;
+  /**
+   * Returns the end timestamp for the last message that was
+   * successfully applied on the publisher node for a given
+   * channel.
+   *
+   * @param[in] Channel ID to check
+   */
+  virtual uint64_t getLastAppliedEndTimestamp(uint32_t channel_id)= 0;
+  /**
+   * Returns the number of channels this publisher
+   * publishes to.
+   */
+  virtual size_t getNumChannels()= 0;
+  /**
+   * Adds to a supplied vector the set of channels this
+   * publisher publishes to.
+   *
+   * @param[out] Pointer to a Vector of channel IDs to add to
+   */
+  virtual void getChannels(std::vector<uint32_t> *out_channels)= 0;
+
+  static bool addPlugin(Publisher *publisher);
+  static void removePlugin(Publisher *publisher);
+private:
+  Publisher();
+  Publisher(const Publisher &);
+  Publisher& operator=(const Publisher &);
+};
+
+} /* namespace plugin */
+} /* namespace drizzled */
+
+#endif /* DRIZZLED_PLUGIN_PUBLISHER_H */

=== modified file 'drizzled/plugin/replication.h'
--- drizzled/plugin/replication.h	2010-03-27 18:03:59 +0000
+++ drizzled/plugin/replication.h	2010-05-05 02:25:42 +0000
@@ -37,7 +37,8 @@
 enum ReplicationReturnCode
 {
   SUCCESS= 0, /* no error */
-  UNKNOWN_ERROR= 1
+  PUBLISHER_ERROR= 1, /* publisher problems...*/
+  UNKNOWN_ERROR= 99
 };
 
 } /* namespace plugin */

=== modified file 'drizzled/plugin/transaction_reader.h'
--- drizzled/plugin/transaction_reader.h	2010-03-05 18:08:49 +0000
+++ drizzled/plugin/transaction_reader.h	2010-05-05 02:25:42 +0000
@@ -25,7 +25,7 @@
 #define DRIZZLED_PLUGIN_TRANSACTION_READER_H
 
 #include "drizzled/plugin/plugin.h"
-#include "drizzled/replication_services.h" /* For global transaction ID typedef */
+#include "drizzled/transaction_services.h" /* For transaction ID typedef */
 
 /**
  * @file Defines the API for a TransactionReader
@@ -66,7 +66,7 @@
    * @retval
    *  false if not found or read successfully
    */
-  virtual bool read(const ReplicationServices::GlobalTransactionId &to_read, 
+  virtual bool read(const TransactionServices::TransactionId &to_read, 
                     message::Transaction *to_fill)= 0;
 };
 

=== modified file 'drizzled/replication_services.cc'
--- drizzled/replication_services.cc	2010-04-05 16:06:01 +0000
+++ drizzled/replication_services.cc	2010-05-05 02:25:42 +0000
@@ -31,13 +31,14 @@
  * - Publisher
  * - Subscriber
  *
- * ReplicationServices is a bridge between replication modules and the kernel,
- * and its primary function is to  */
+ * ReplicationServices is a bridge between replication modules and the kernel
+ */
 
 #include "config.h"
 #include "drizzled/replication_services.h"
 #include "drizzled/plugin/transaction_replicator.h"
 #include "drizzled/plugin/transaction_applier.h"
+#include "drizzled/plugin/publisher.h"
 #include "drizzled/message/transaction.pb.h"
 #include "drizzled/gettext.h"
 #include "drizzled/session.h"
@@ -45,6 +46,7 @@
 
 #include <string>
 #include <vector>
+#include <functional>
 #include <algorithm>
 
 using namespace std;
@@ -140,6 +142,16 @@
   return true;
 }
 
+void ReplicationServices::attachPublisher(plugin::Publisher *in_publisher)
+{
+  publishers.push_back(in_publisher);
+}
+
+void ReplicationServices::detachPublisher(plugin::Publisher *in_publisher)
+{
+  publishers.erase(std::find(publishers.begin(), publishers.end(), in_publisher));
+}
+
 void ReplicationServices::attachReplicator(plugin::TransactionReplicator *in_replicator)
 {
   replicators.push_back(in_replicator);
@@ -178,22 +190,18 @@
 
     result= cur_repl->replicate(cur_appl, in_session, to_push);
 
-    if (result == plugin::SUCCESS)
-    {
-      /* 
-       * We update the timestamp for the last applied Transaction so that
-       * publisher plugins can ask the replication services when the
-       * last known applied Transaction was using the getLastAppliedTimestamp()
-       * method.
-       */
-      last_applied_timestamp.fetch_and_store(to_push.transaction_context().end_timestamp());
-    }
-    else
+    if (result != plugin::SUCCESS)
       return result;
   }
+
   return result;
 }
 
+ReplicationServices::Publishers &ReplicationServices::getPublishers()
+{
+  return publishers;
+}
+
 ReplicationServices::ReplicationStreams &ReplicationServices::getReplicationStreams()
 {
   return replication_streams;

=== modified file 'drizzled/replication_services.h'
--- drizzled/replication_services.h	2010-04-05 16:06:01 +0000
+++ drizzled/replication_services.h	2010-05-05 02:25:42 +0000
@@ -43,6 +43,8 @@
 {
   class TransactionReplicator;
   class TransactionApplier;
+  class Subscriber;
+  class Publisher;
 }
 namespace message
 {
@@ -50,14 +52,29 @@
 }
 
 /**
- * This is a class which manages transforming internal 
- * transactional events into GPB messages and sending those
- * events out through registered replicators and appliers.
+ * This is a class which manages communication between publishers
+ * and subscribers and sends transaction messages out to
+ * replicator and applier plugins.
+ *
+ * The relationship betweem the above classes is as follows:
+ *
+ * TransactionReplicators are responsible for filtering or transforming
+ * Transaction messages before passing them off to TransactionAppliers.
+ *
+ * TransactionAppliers "apply" the Transaction messages to something --
+ * this "something" could be anything, for instance a log file or a message queue.
+ *
+ * Publishers are responsible for publishing information
+ * about the Transaction messages which have been applied to 
+ * a particular channel.
+ *
+ * Subscribers are responsible for asking a Publisher for a data
+ * stream of Transaction messages that it will apply to its own
+ * data store.
  */
 class ReplicationServices
 {
 public:
-  typedef uint64_t GlobalTransactionId;
   /**
    * Types of messages that can go in the transaction
    * log file.  Every time something is written into the
@@ -69,8 +86,18 @@
     TRANSACTION= 1, /* A GPB Transaction Message */
     BLOB= 2 /* A BLOB value */
   };
-  typedef std::pair<plugin::TransactionReplicator *, plugin::TransactionApplier *> ReplicationPair;
-  typedef std::vector<ReplicationPair> ReplicationStreams;
+  /**
+   * A replication stream is the pairing of a replicator
+   * plugin with an applied plugin.
+   */
+  typedef std::pair<plugin::TransactionReplicator *, plugin::TransactionApplier *> ReplicationStream;
+  typedef std::vector<ReplicationStream> ReplicationStreams;
+  /**
+   * A publisher channel is the pairing of a publisher plugin
+   * with a channel ID.
+   */
+  typedef std::vector<uint32_t> PublisherChannels;
+  typedef std::vector<plugin::Publisher *> Publishers;
   /**
    * Method which is called after plugins have been loaded but
    * before the first client connects.  It determines if the registration
@@ -84,6 +111,7 @@
    * tracking...
    */
   bool evaluateRegisteredPlugins();
+  typedef std::vector<plugin::Subscriber *> Subscribers;
   /** 
    * Helper method which pushes a constructed message out to the registered
    * replicator and applier plugins.
@@ -116,11 +144,32 @@
   bool isActive() const;
 
   /**
+   * Returns the list of publishers this node knows about
+   */
+  Publishers &getPublishers();
+
+  /**
    * Returns the list of replication streams
    */
   ReplicationStreams &getReplicationStreams();
 
   /**
+   * Attaches a publisher to our internal collection of
+   * publishers.
+   *
+   * @param Pointer to a publisher to attach/register
+   */
+  void attachPublisher(plugin::Publisher *in_publisher);
+  
+  /**
+   * Detaches/unregisters a publisher with our internal
+   * collection of publishers.
+   *
+   * @param Pointer to the publisher to detach
+   */
+  void detachPublisher(plugin::Publisher *in_publisher);
+
+  /**
    * Attaches a replicator to our internal collection of
    * replicators.
    *
@@ -152,36 +201,36 @@
    * @param Pointer to the applier to detach
    */
   void detachApplier(plugin::TransactionApplier *in_applier);
-
-  /** 
-   * Returns the timestamp of the last Transaction which was sent to an
-   * applier.
-   */
-  uint64_t getLastAppliedTimestamp() const;
 private:
   typedef std::vector<plugin::TransactionReplicator *> Replicators;
   typedef std::vector<std::pair<std::string, plugin::TransactionApplier *> > Appliers;
   /** 
-   * Atomic boolean set to true if any *active* replicators
-   * or appliers are actually registered.
+   * Set to true if there are replication streams
    */
   bool is_active;
-  /**
-   * The timestamp of the last time a Transaction message was successfully
-   * applied (sent to an Applier)
-   */
-  atomic<uint64_t> last_applied_timestamp;
   /** Our collection of registered replicator plugins */
   Replicators replicators;
   /** Our collection of registered applier plugins and their requested replicator plugin names */
   Appliers appliers;
   /** Our replication streams */
   ReplicationStreams replication_streams;
+  /** The set of known publishers */
+  Publishers publishers;
+  /** The set of known subscribers */
+  Subscribers subscribers;
   /**
    * Strips underscores and lowercases supplied replicator name
    * or requested name, and appends the suffix "replicator" if missing...
    */
   void normalizeReplicatorName(std::string &name);
+  /**
+   * Helper method which is called after any change in the
+   * registered appliers or replicators to evaluate whether
+   * any remaining plugins are actually active.
+   * 
+   * This method properly sets the is_active member variable.
+   */
+  void evaluateActivePlugins();
 };
 
 } /* namespace drizzled */

=== modified file 'drizzled/transaction_services.h'
--- drizzled/transaction_services.h	2010-03-31 19:04:12 +0000
+++ drizzled/transaction_services.h	2010-05-05 02:25:42 +0000
@@ -40,6 +40,7 @@
 }
 
 class Session;
+class Table;
 class NamedSavepoint;
 
 /**

=== modified file 'plugin/replication_dictionary/module.cc'
--- plugin/replication_dictionary/module.cc	2010-04-05 16:06:01 +0000
+++ plugin/replication_dictionary/module.cc	2010-05-05 02:25:42 +0000
@@ -31,11 +31,15 @@
 
 #include "config.h"
 #include "streams.h"
+#include "publishers.h"
+#include "publisher_channels.h"
 
 using namespace std;
 using namespace drizzled;
 
 static ReplicationStreamsTool *streams_tool;
+static PublishersTool *publishers_tool;
+static PublisherChannelsTool *publisher_channels_tool;
 
 static int init(plugin::Context &context)
 {
@@ -44,13 +48,40 @@
   if (streams_tool != NULL)
   {
     context.add(streams_tool);
-    return 0;
-  }
-  else
-  {
-    // error?
-    return 1;
-  }
+  }
+  else
+  {
+    // error?
+    return 1;
+  }
+
+  publishers_tool= new (nothrow) PublishersTool;
+
+  if (publishers_tool != NULL)
+  {
+    context.add(publishers_tool);
+  }
+  else
+  {
+    delete streams_tool;
+    // error?
+    return 1;
+  }
+
+  publisher_channels_tool= new (nothrow) PublisherChannelsTool;
+
+  if (publisher_channels_tool != NULL)
+  {
+    context.add(publisher_channels_tool);
+  }
+  else
+  {
+    delete streams_tool;
+    delete publishers_tool;
+    // error?
+    return 1;
+  }
+  return 0;
 }
 
 DRIZZLE_PLUGIN(init, NULL);

=== modified file 'plugin/replication_dictionary/plugin.ini'
--- plugin/replication_dictionary/plugin.ini	2010-04-11 15:50:01 +0000
+++ plugin/replication_dictionary/plugin.ini	2010-05-05 02:25:42 +0000
@@ -1,5 +1,5 @@
 [plugin]
-title=Registry Dictionary
+title=Replication Dictionary
 author=Jay Pipes <[e-mail]>
 license=PLUGIN_LICENSE_BSD
 description=Provides dictionary tables for replication system
@@ -7,6 +7,10 @@
 sources=
   module.cc
   streams.cc
+  publishers.cc
+  publisher_channels.cc
 headers=
   streams.h
+  publishers.h
+  publisher_channels.h
 static=yes

=== added file 'plugin/replication_dictionary/publisher_channels.cc'
--- plugin/replication_dictionary/publisher_channels.cc	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/publisher_channels.cc	2010-05-05 02:25:42 +0000
@@ -0,0 +1,105 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ * Copyright (c) 2010 Jay Pipes <[e-mail]>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright notice,
+ *     this list of conditions and the following disclaimer in the documentation
+ *     and/or other materials provided with the distribution.
+ *   * Neither the name of Drizzle nor the names of its contributors
+ *     may be used to endorse or promote products derived from this software
+ *     without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
+ * THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "config.h"
+#include "publisher_channels.h"
+
+#include <drizzled/plugin/publisher.h>
+
+using namespace std;
+using namespace drizzled;
+
+PublisherChannelsTool::PublisherChannelsTool() :
+  plugin::TableFunction("DATA_DICTIONARY", "REPLICATION_PUBLISHER_CHANNELS")
+{
+  add_field("PUBLISHER");
+  add_field("CHANNEL_ID", plugin::TableFunction::NUMBER);
+  add_field("LAST_APPLIED_TRANSACTION_ID", plugin::TableFunction::NUMBER);
+  add_field("LAST_APPLIED_END_TIMESTAMP", plugin::TableFunction::NUMBER);
+}
+
+PublisherChannelsTool::Generator::Generator(Field **arg) :
+  plugin::TableFunction::Generator(arg)
+{
+  ReplicationServices &replication_services= ReplicationServices::singleton();
+  ReplicationServices::Publishers &publishers= replication_services.getPublishers();
+
+  /*
+   * Fill our row vector with information from each
+   * publisher and the publisher's collection of
+   * apply information for each channel
+   */
+  vector<uint32_t> channels;
+  for (ReplicationServices::Publishers::iterator iter= publishers.begin();
+       iter != publishers.end();
+       ++iter)
+  {
+    plugin::Publisher *publisher= *iter;
+    channels.clear();
+    channels.reserve(publisher->getNumChannels());
+    publisher->getChannels(&channels);
+    for (vector<uint32_t>::iterator channel_iter= channels.begin();
+         channel_iter != channels.end();
+         ++channel_iter)
+    {
+      rows.push_back(make_pair(publisher, *channel_iter));
+    }
+  }
+
+  it= rows.begin();
+  end= rows.end();
+}
+
+bool PublisherChannelsTool::Generator::populate()
+{
+  if (it == end)
+    return false;
+
+  plugin::Publisher *publisher= (*it).first;
+  uint32_t channel= (*it).second;
+
+  /* PUBLISHER */
+  push(publisher->getName());
+
+  /* CHANNEL_ID */
+  push(static_cast<uint64_t>(channel));
+
+  /* LAST_APPLIED_TRANSACTION_ID */
+  push(static_cast<uint64_t>(publisher->getLastAppliedTransactionId(channel)));
+
+  /* LAST_APPLIED_END_TIMESTAMP */
+  push(static_cast<uint64_t>(publisher->getLastAppliedEndTimestamp(channel)));
+
+  ++it;
+
+  return true;
+}
+

=== added file 'plugin/replication_dictionary/publisher_channels.h'
--- plugin/replication_dictionary/publisher_channels.h	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/publisher_channels.h	2010-05-05 02:25:42 +0000
@@ -0,0 +1,67 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ * Copyright (c) 2010 Jay Pipes <[e-mail]>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright notice,
+ *     this list of conditions and the following disclaimer in the documentation
+ *     and/or other materials provided with the distribution.
+ *   * Neither the name of Drizzle nor the names of its contributors
+ *     may be used to endorse or promote products derived from this software
+ *     without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
+ * THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef PLUGIN_REPLICATION_DICTIONARY_PUBLISHER_CHANNELS_H
+#define PLUGIN_REPLICATION_DICTIONARY_PUBLISHER_CHANNELS_H
+
+#include <drizzled/plugin/table_function.h>
+#include <drizzled/replication_services.h>
+
+#include <utility>
+#include <vector>
+
+class PublisherChannelsTool : public drizzled::plugin::TableFunction
+{
+public:
+  PublisherChannelsTool();
+
+  class Generator : public drizzled::plugin::TableFunction::Generator 
+  {
+  public:
+    Generator(drizzled::Field **arg);
+
+    bool populate();
+  private:
+    typedef std::pair<drizzled::plugin::Publisher *,
+                      uint32_t /* index into apply_info in publisher manifest */> Row;
+    typedef std::vector<Row> Rows;
+    Rows rows;
+    Rows::iterator it;
+    Rows::iterator end;
+  };
+
+  Generator *generator(drizzled::Field **arg)
+  {
+    return new Generator(arg);
+  }
+};
+
+#endif /* PLUGIN_REPLICATION_DICTIONARY_PUBLISHER_CHANNELS_H */

=== added file 'plugin/replication_dictionary/publishers.cc'
--- plugin/replication_dictionary/publishers.cc	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/publishers.cc	2010-05-05 02:25:42 +0000
@@ -0,0 +1,71 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ * Copyright (c) 2010 Jay Pipes <[e-mail]>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright notice,
+ *     this list of conditions and the following disclaimer in the documentation
+ *     and/or other materials provided with the distribution.
+ *   * Neither the name of Drizzle nor the names of its contributors
+ *     may be used to endorse or promote products derived from this software
+ *     without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
+ * THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "config.h"
+#include "publishers.h"
+
+#include <drizzled/replication_services.h>
+#include <drizzled/plugin/publisher.h>
+
+using namespace std;
+using namespace drizzled;
+
+PublishersTool::PublishersTool() :
+  plugin::TableFunction("DATA_DICTIONARY", "REPLICATION_PUBLISHERS")
+{
+  add_field("PUBLISHER_NAME");
+}
+
+PublishersTool::Generator::Generator(Field **arg) :
+  plugin::TableFunction::Generator(arg)
+{
+  ReplicationServices &replication_services= ReplicationServices::singleton();
+  ReplicationServices::Publishers &publishers= replication_services.getPublishers();
+
+  it= publishers.begin();
+  end= publishers.end();
+}
+
+bool PublishersTool::Generator::populate()
+{
+  if (it == end)
+    return false;
+
+  plugin::Publisher *publisher= *it;
+
+  /* PUBLISHER */
+  push(publisher->getName());
+
+  ++it;
+
+  return true;
+}
+

=== added file 'plugin/replication_dictionary/publishers.h'
--- plugin/replication_dictionary/publishers.h	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/publishers.h	2010-05-05 02:25:42 +0000
@@ -0,0 +1,60 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ * Copyright (c) 2010 Jay Pipes <[e-mail]>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright notice,
+ *     this list of conditions and the following disclaimer in the documentation
+ *     and/or other materials provided with the distribution.
+ *   * Neither the name of Drizzle nor the names of its contributors
+ *     may be used to endorse or promote products derived from this software
+ *     without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
+ * THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef PLUGIN_REPLICATION_DICTIONARY_PUBLISHERS_H
+#define PLUGIN_REPLICATION_DICTIONARY_PUBLISHERS_H
+
+#include <drizzled/plugin/table_function.h>
+#include <drizzled/replication_services.h>
+
+class PublishersTool : public drizzled::plugin::TableFunction
+{
+public:
+  PublishersTool();
+
+  class Generator : public drizzled::plugin::TableFunction::Generator 
+  {
+    drizzled::ReplicationServices::Publishers::iterator it;
+    drizzled::ReplicationServices::Publishers::iterator end;
+
+  public:
+    Generator(drizzled::Field **arg);
+
+    bool populate();
+  };
+
+  Generator *generator(drizzled::Field **arg)
+  {
+    return new Generator(arg);
+  }
+};
+
+#endif /* PLUGIN_REPLICATION_DICTIONARY_PUBLISHERS_H */

=== modified file 'plugin/replication_dictionary/tests/r/data_dictionary.result'
--- plugin/replication_dictionary/tests/r/data_dictionary.result	2010-04-05 16:11:15 +0000
+++ plugin/replication_dictionary/tests/r/data_dictionary.result	2010-05-05 02:25:42 +0000
@@ -2,9 +2,28 @@
 SELECT count(*) FROM replication_streams;
 count(*)
 #
+SELECT count(*) FROM replication_publishers;
+count(*)
+#
+SELECT count(*) FROM replication_publisher_channels;
+count(*)
+#
 show create table replication_streams;
 Table	Create Table
 replication_streams	CREATE TABLE `replication_streams` (
   `REPLICATOR` varchar(64) NOT NULL DEFAULT '',
   `APPLIER` varchar(64) NOT NULL DEFAULT ''
 ) ENGINE=FunctionEngine
+show create table replication_publishers;
+Table	Create Table
+replication_publishers	CREATE TABLE `replication_publishers` (
+  `PUBLISHER_NAME` varchar(64) NOT NULL DEFAULT ''
+) ENGINE=FunctionEngine
+show create table replication_publisher_channels;
+Table	Create Table
+replication_publisher_channels	CREATE TABLE `replication_publisher_channels` (
+  `PUBLISHER` varchar(64) NOT NULL DEFAULT '',
+  `CHANNEL_ID` bigint NOT NULL DEFAULT '0',
+  `LAST_APPLIED_TRANSACTION_ID` bigint NOT NULL DEFAULT '0',
+  `LAST_APPLIED_END_TIMESTAMP` bigint NOT NULL DEFAULT '0'
+) ENGINE=FunctionEngine

=== added file 'plugin/replication_dictionary/tests/r/replication_publisher_channels.result'
--- plugin/replication_dictionary/tests/r/replication_publisher_channels.result	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/tests/r/replication_publisher_channels.result	2010-05-05 02:25:42 +0000
@@ -0,0 +1,20 @@
+use data_dictionary;
+SET @@global.transaction_log_truncate_debug= 1;
+SELECT * FROM data_dictionary.replication_publisher_channels;
+PUBLISHER	CHANNEL_ID	LAST_APPLIED_TRANSACTION_ID	LAST_APPLIED_END_TIMESTAMP
+transaction_log_publisher	1	0	0
+use test;
+DROP TABLE IF EXISTS t1;
+SET @@global.transaction_log_truncate_debug= 1;
+CREATE TABLE t1
+(
+id INT NOT NULL PRIMARY KEY
+, padding VARCHAR(20) NOT NULL
+);
+INSERT INTO t1 VALUES (1, 'I love testing');
+INSERT INTO t1 VALUES (2, 'I hate testing');
+DROP TABLE t1;
+use data_dictionary;
+SELECT * FROM data_dictionary.replication_publisher_channels;
+PUBLISHER	CHANNEL_ID	LAST_APPLIED_TRANSACTION_ID	LAST_APPLIED_END_TIMESTAMP
+transaction_log_publisher	1	4	X

=== added file 'plugin/replication_dictionary/tests/r/replication_publishers.result'
--- plugin/replication_dictionary/tests/r/replication_publishers.result	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/tests/r/replication_publishers.result	2010-05-05 02:25:42 +0000
@@ -0,0 +1,4 @@
+use data_dictionary;
+SELECT * FROM data_dictionary.replication_publishers;
+PUBLISHER_NAME
+transaction_log_publisher

=== modified file 'plugin/replication_dictionary/tests/t/data_dictionary.test'
--- plugin/replication_dictionary/tests/t/data_dictionary.test	2010-04-05 16:06:01 +0000
+++ plugin/replication_dictionary/tests/t/data_dictionary.test	2010-05-05 02:25:42 +0000
@@ -5,4 +5,12 @@
 --replace_column 1 #
 SELECT count(*) FROM replication_streams;
 
+--replace_column 1 #
+SELECT count(*) FROM replication_publishers;
+
+--replace_column 1 #
+SELECT count(*) FROM replication_publisher_channels;
+
 show create table replication_streams;
+show create table replication_publishers;
+show create table replication_publisher_channels;

=== added symlink 'plugin/replication_dictionary/tests/t/replication_publisher_channels-master.opt'
=== target is u'startup-master.opt'
=== added file 'plugin/replication_dictionary/tests/t/replication_publisher_channels.test'
--- plugin/replication_dictionary/tests/t/replication_publisher_channels.test	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/tests/t/replication_publisher_channels.test	2010-05-05 02:25:42 +0000
@@ -0,0 +1,31 @@
+# Drizzle's replication dictionary's replication_publisher_channels table
+
+use data_dictionary;
+
+SET @@global.transaction_log_truncate_debug= 1;
+
+SELECT * FROM data_dictionary.replication_publisher_channels;
+
+use test;
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+SET @@global.transaction_log_truncate_debug= 1;
+
+CREATE TABLE t1
+(
+  id INT NOT NULL PRIMARY KEY
+, padding VARCHAR(20) NOT NULL
+);
+
+INSERT INTO t1 VALUES (1, 'I love testing');
+INSERT INTO t1 VALUES (2, 'I hate testing');
+
+DROP TABLE t1;
+
+use data_dictionary;
+
+--replace_column 4 X
+SELECT * FROM data_dictionary.replication_publisher_channels;

=== added symlink 'plugin/replication_dictionary/tests/t/replication_publishers-master.opt'
=== target is u'startup-master.opt'
=== added file 'plugin/replication_dictionary/tests/t/replication_publishers.test'
--- plugin/replication_dictionary/tests/t/replication_publishers.test	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/tests/t/replication_publishers.test	2010-05-05 02:25:42 +0000
@@ -0,0 +1,5 @@
+# Drizzle's replication dictionary's replication_publishers table.
+
+use data_dictionary;
+
+SELECT * FROM data_dictionary.replication_publishers;

=== added symlink 'plugin/replication_dictionary/tests/t/replication_streams-master.opt'
=== target is u'startup-master.opt'
=== removed file 'plugin/replication_dictionary/tests/t/replication_streams-master.opt'
--- plugin/replication_dictionary/tests/t/replication_streams-master.opt	2010-04-05 16:06:01 +0000
+++ plugin/replication_dictionary/tests/t/replication_streams-master.opt	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
---transaction-log-enable --transaction-log-use-replicator=filtered

=== added symlink 'plugin/replication_dictionary/tests/t/slap-master.opt'
=== target is u'startup-master.opt'
=== added file 'plugin/replication_dictionary/tests/t/startup-master.opt'
--- plugin/replication_dictionary/tests/t/startup-master.opt	1970-01-01 00:00:00 +0000
+++ plugin/replication_dictionary/tests/t/startup-master.opt	2010-05-05 02:25:42 +0000
@@ -0,0 +1,1 @@
+--transaction-log-enable --transaction-log-use-replicator=filtered

=== modified file 'plugin/schema_dictionary/tests/r/data_dictionary.result'
--- plugin/schema_dictionary/tests/r/data_dictionary.result	2010-04-07 21:54:30 +0000
+++ plugin/schema_dictionary/tests/r/data_dictionary.result	2010-05-05 02:25:42 +0000
@@ -466,6 +466,11 @@
 #
 #
 #
+#
+#
+#
+#
+#
 SELECT count(*) FROM REFERENTIAL_CONSTRAINTS;
 count(*)
 #

=== modified file 'plugin/transaction_log/background_worker.cc'
--- plugin/transaction_log/background_worker.cc	2010-02-04 08:14:46 +0000
+++ plugin/transaction_log/background_worker.cc	2010-05-05 02:25:42 +0000
@@ -60,7 +60,6 @@
 #include <drizzled/gettext.h>
 #include <drizzled/errmsg_print.h>
 
-
 #include "transaction_log.h"
 #include "background_worker.h"
 

=== modified file 'plugin/transaction_log/module.cc'
--- plugin/transaction_log/module.cc	2010-04-12 19:15:00 +0000
+++ plugin/transaction_log/module.cc	2010-05-05 02:25:42 +0000
@@ -39,6 +39,7 @@
 #include "print_transaction_message.h"
 #include "hexdump_transaction_message.h"
 #include "background_worker.h"
+#include "publisher.h"
 
 #include <drizzled/plugin/plugin.h>
 #include <drizzled/session.h>
@@ -101,6 +102,8 @@
 extern TransactionLogIndex *transaction_log_index;
 /** Transaction Log descriptor defined in transaction_log.cc */
 extern TransactionLog *transaction_log;
+/** Publisher plugin defined in publisher.cc */
+extern TransactionLogPublisher *transaction_log_publisher;
 /** Transaction Log descriptor defined in transaction_log.cc */
 extern TransactionLogApplier *transaction_log_applier;
 
@@ -153,10 +156,23 @@
       }
     }
 
+    /* Create and initialize the transaction log publisher */
+    try
+    {
+      transaction_log_publisher= new TransactionLogPublisher("transaction_log_publisher");
+    }
+    catch (exception &e)
+    {
+      errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to create the TransactionLogPublisher.  Got error: %s\n"), 
+                    e.what());
+      return 1;
+    }
+
     /* Create the applier plugin and register it */
     transaction_log_applier= new (nothrow) TransactionLogApplier("transaction_log_applier",
                                                                  transaction_log, 
                                                                  transaction_log_index, 
+                                                                 transaction_log_publisher, 
                                                                  sysvar_transaction_log_num_write_buffers);
     if (transaction_log_applier == NULL)
     {
@@ -165,6 +181,15 @@
       return 1;
     }
     context.add(transaction_log_applier);
+    context.add(transaction_log_publisher);
+
+    /* 
+     * Tell the replication services component about our applier 
+     *
+     * @note This is a hack to allow us to pair a replicator with
+     * an applied.  Should be refactored to not rely on a call to the
+     * replication services here...
+     */
     ReplicationServices &replication_services= ReplicationServices::singleton();
     string replicator_name(sysvar_transaction_log_use_replicator);
     replication_services.attachApplier(transaction_log_applier, replicator_name);
@@ -213,6 +238,7 @@
     {
       transaction_log->truncate();
       transaction_log_index->clear();
+      transaction_log_publisher->clear();
     }
   }
 }

=== added file 'plugin/transaction_log/plugin.am'
--- plugin/transaction_log/plugin.am	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/plugin.am	2010-05-05 02:25:42 +0000
@@ -0,0 +1,27 @@
+# vim:ft=automake
+# Copyright (C) 2010 Jay Pipes
+#
+# 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+
+EXTRA_DIST += plugin/transaction_log/replication.proto
+
+BUILT_SOURCES += plugin/transaction_log/replication.pb.h plugin/transaction_log/replication.pb.cc
+
+CLEANFILES += plugin/transaction_log/replication.pb.h plugin/transaction_log/replication.pb.cc
+
+TRX_LOG_PROTO_PATH=${top_builddir}/plugin/transaction_log/:${top_srcdir}/plugin/transaction_log/
+
+plugin/transaction_log/replication.pb.cc plugin/transaction_log/replication.pb.h: plugin/transaction_log/replication.proto
+	$(PROTOC) --proto_path=${TRX_LOG_PROTO_PATH} \
+		--cpp_out=${top_builddir}/plugin/transaction_log/ $<

=== modified file 'plugin/transaction_log/plugin.ini'
--- plugin/transaction_log/plugin.ini	2010-03-27 18:51:59 +0000
+++ plugin/transaction_log/plugin.ini	2010-05-05 02:25:42 +0000
@@ -6,8 +6,8 @@
 title=Transaction Log
 description=Log of Transaction Messages
 load_by_default=yes
-sources= background_worker.cc hexdump_transaction_message.cc module.cc print_transaction_message.cc transaction_log.cc transaction_log_applier.cc transaction_log_entry.cc transaction_log_index.cc transaction_log_reader.cc data_dictionary_schema.cc write_buffer.cc
-headers= background_worker.h hexdump_transaction_message.h print_transaction_message.h transaction_log.h transaction_log_applier.h transaction_log_entry.h transaction_log_index.h transaction_log_reader.h data_dictionary_schema.h write_buffer.h
+sources= background_worker.cc hexdump_transaction_message.cc module.cc print_transaction_message.cc transaction_log.cc transaction_log_applier.cc transaction_log_entry.cc transaction_log_index.cc transaction_log_reader.cc publisher.cc data_dictionary_schema.cc write_buffer.cc replication.pb.cc
+headers= background_worker.h hexdump_transaction_message.h print_transaction_message.h transaction_log.h transaction_log_applier.h transaction_log_entry.h transaction_log_index.h transaction_log_reader.h publisher.h data_dictionary_schema.h write_buffer.h replication.pb.h
 libs=${top_builddir}/drizzled/algorithm/libhash.la 
 libadd=$(LIBZ)
-cxxflags=${PROTOSKIP_WARNINGS}
+cxxflags=${PROTOSKIP_WARNINGS} ${NO_WERROR}

=== added file 'plugin/transaction_log/publisher.cc'
--- plugin/transaction_log/publisher.cc	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/publisher.cc	2010-05-05 02:25:42 +0000
@@ -0,0 +1,213 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (C) 2008-2009 Sun Microsystems
+ *  Copyright (c) 2010 Jay Pipes <[e-mail]>
+ *
+ *  Authors:
+ *
+ *    Jay Pipes <[e-mail]>
+ *
+ *  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 2 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, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+/**
+ * @file
+ *
+ * Defines the implementation of the publisher plugin
+ * for the transaction log.
+ *
+ * @see drizzled/plugin/publisher.h
+ * @see drizzled/plugin/subscriber.h
+ */
+
+#include "config.h"
+#include "publisher.h"
+
+#include <fcntl.h>
+#include <unistd.h>
+#include <string>
+#include <utility>
+#include <stdexcept>
+
+#include <drizzled/internal/my_sys.h> /* for internal::my_sync */
+#include <drizzled/gettext.h>
+#include <drizzled/errmsg_print.h>
+
+using namespace std;
+using namespace drizzled;
+
+TransactionLogPublisher *transaction_log_publisher= NULL; /* The singleton transaction log publisher */
+
+const std::string TransactionLogPublisher::MANIFEST_FILE_NAME= ("publisher.manifest"); 
+
+TransactionLogPublisher::TransactionLogPublisher(string name_arg) :
+  plugin::Publisher(name_arg),
+  publisher_manifest(),
+  channel_apply_map(),
+  manifest_file(-1)
+{
+  pthread_mutex_init(&latch, NULL);
+
+  if (not readManifestFile())
+  {
+    throw runtime_error(_("Failed to read publisher's manifest file")); 
+  }
+
+  manifest_file= open(MANIFEST_FILE_NAME.c_str(), O_CREAT|O_SYNC|O_WRONLY, S_IRWXU);
+}
+
+TransactionLogPublisher::~TransactionLogPublisher()
+{
+  try
+  {
+    (void) writeManifestFile();
+    (void) syncManifestFile();
+    close(manifest_file);
+  }
+  catch (...)
+  {} /* enforce nothrow guarantee on destructors... */
+  pthread_mutex_destroy(&latch);
+}
+
+bool TransactionLogPublisher::readManifestFile()
+{
+  PublisherManifest &pm= getPublisherManifest();
+  int read_file= open(MANIFEST_FILE_NAME.c_str(), O_RDONLY);
+  if (read_file == -1)
+  {
+    /* 
+     * If file doesn't exist, just create a default manifest message
+     * in memory.  This message will be written to disk into the
+     * manifest file later.
+     */
+    if (errno == ENOENT)
+    {
+      /* Create a default manifest with channel 1 and no applied transactions */
+      PublisherManifest::PublisherApplyInfo *appl_info= pm.add_apply_info();
+      appl_info->set_channel_id(1);
+      appl_info->set_last_applied_transaction_id(0);
+      appl_info->set_last_applied_end_timestamp(0);
+    }
+    else
+    {
+      errmsg_printf(ERRMSG_LVL_ERROR, _("Unable to read publisher's manifest file!  Got error: %s\n"), strerror(errno));
+      close(read_file);
+      return false;
+    }
+  }
+  else
+  {
+    if (not pm.ParseFromFileDescriptor(read_file))
+    {
+      errmsg_printf(ERRMSG_LVL_ERROR,
+                    _("Unable to parse publisher's manifest file!\n"
+                      "PublisherManifest was initialized, but contains corrupted data:\n%s\n"),
+                    pm.DebugString().c_str());
+      close(read_file);
+      return false;
+    }
+  }
+  uint32_t num_channels= pm.apply_info_size();
+  for (uint32_t x= 0; x < num_channels; ++x)
+  {
+    PublisherManifest::PublisherApplyInfo *apply_info= pm.mutable_apply_info(x);
+    channel_apply_map.insert(make_pair(apply_info->channel_id(), apply_info));
+  }
+  close(read_file);
+  return true;
+}
+
+void TransactionLogPublisher::clear()
+{
+  publisher_manifest.Clear();
+  /* Create a default manifest with channel 1 and no applied transactions */
+  PublisherManifest::PublisherApplyInfo *appl_info= publisher_manifest.add_apply_info();
+  appl_info->set_channel_id(1);
+  appl_info->set_last_applied_transaction_id(0);
+  appl_info->set_last_applied_end_timestamp(0);
+  (void) ftruncate(manifest_file, 0);
+  (void) internal::my_sync(manifest_file, 0);
+}
+
+void TransactionLogPublisher::setLastApplied(const message::Transaction &applied)
+{
+  uint32_t channel_applied;
+  if (applied.transaction_context().has_channel_id())
+    channel_applied= applied.transaction_context().channel_id();
+  else
+    channel_applied= 1;
+  lock();
+  channel_apply_map[channel_applied]->set_last_applied_transaction_id(applied.transaction_context().transaction_id());
+  channel_apply_map[channel_applied]->set_last_applied_end_timestamp(applied.transaction_context().end_timestamp());
+  unlock();
+  (void) writeManifestFile();
+}
+
+TransactionServices::TransactionId TransactionLogPublisher::getLastAppliedTransactionId(uint32_t channel_id)
+{
+  lock();
+  TransactionServices::TransactionId result= channel_apply_map[channel_id]->last_applied_transaction_id();
+  unlock();
+  return result;
+}
+
+uint64_t TransactionLogPublisher::getLastAppliedEndTimestamp(uint32_t channel_id)
+{
+  lock();
+  uint64_t result= channel_apply_map[channel_id]->last_applied_end_timestamp();
+  unlock();
+  return result;
+}
+
+bool TransactionLogPublisher::writeManifestFile()
+{
+  /* write the manifest message out to our file... */
+  PublisherManifest &pm= getPublisherManifest();
+  string write_buffer("");
+  lock();
+  bool result= pm.SerializeToString(&write_buffer);
+  if (unlikely(not result))
+  {
+    errmsg_printf(ERRMSG_LVL_ERROR,
+                  _("Unable to write publisher's manifest file!\n"
+                    "PublisherManifest contains data:\n%s\n"),
+                  pm.DebugString().c_str());
+    unlock();
+    return false;
+  }
+  uint32_t num_bytes_to_write= write_buffer.size();
+  ssize_t written= pwrite(manifest_file, write_buffer.c_str(), num_bytes_to_write, 0);
+  result= (written != static_cast<ssize_t>(num_bytes_to_write));
+  unlock();
+  return result;
+}
+
+bool TransactionLogPublisher::syncManifestFile()
+{
+  return internal::my_sync(manifest_file, 0) == 0;
+}
+
+void TransactionLogPublisher::getChannels(vector<uint32_t> *out_channels)
+{
+  lock();
+  PublisherManifest &pm= getPublisherManifest();
+  size_t num_channels= pm.apply_info_size();
+  for (uint32_t x= 0; x < num_channels; ++x)
+  {
+    out_channels->push_back(pm.apply_info(x).channel_id());
+  }
+  unlock();
+}

=== added file 'plugin/transaction_log/publisher.h'
--- plugin/transaction_log/publisher.h	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/publisher.h	2010-05-05 02:25:42 +0000
@@ -0,0 +1,161 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (c) 2010 Jay Pipes <[e-mail]>
+ *
+ *  Authors:
+ *
+ *  Jay Pipes <[e-mail]>
+ *
+ *  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 2 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, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+/**
+ * @file
+ *
+ * Defines the API of the transaction log's publisher plugin.
+ *
+ * @see drizzled/plugin/publisher.h
+ * @see drizzled/plugin/subscriber.h
+ */
+
+#ifndef PLUGIN_TRANSACTION_LOG_PUBLISHER_H
+#define PLUGIN_TRANSACTION_LOG_PUBLISHER_H
+
+#include <drizzled/plugin/publisher.h>
+
+#include "replication.pb.h"
+
+#include <drizzled/unordered_map.h>
+#include <string>
+#include <pthread.h>
+
+/**
+ * The publisher class for the transaction log.  This class is
+ * responsible mostly for keeping track of the last applied
+ * transaction ID and timestamps on this node for each channel,
+ * and for communicating with subscribers of transaction log data.
+ */
+class TransactionLogPublisher: public drizzled::plugin::Publisher
+{
+public:
+  /** Constructor */
+  TransactionLogPublisher(std::string name_arg);
+  /** Destructor */
+  ~TransactionLogPublisher();
+  /**
+   * Returns the transaction ID for the last message that was
+   * successfully applied on the publisher node for a given
+   * channel.
+   *
+   * @param[in] Channel ID to check
+   */
+  drizzled::TransactionServices::TransactionId getLastAppliedTransactionId(uint32_t channel_id);
+  /**
+   * Returns the end timestamp for the last message that was
+   * successfully applied on the publisher node for a given
+   * channel.
+   *
+   * @param[in] Channel ID to check
+   */
+  uint64_t getLastAppliedEndTimestamp(uint32_t channel_id);
+  /**
+   * Sets the last transaction ID and timestamp based
+   * on supplied transaction message
+   *
+   * @param[in] Transaction message to check for ID and timestamp
+   */
+  void setLastApplied(const drizzled::message::Transaction &applied);
+  /**
+   * Simple debugging method which sets each channel's applied
+   * transaction ID and end timestamp to zero.
+   */
+  void clear();
+  /**
+   * Returns the number of channels this publisher
+   * publishes to.
+   */
+  size_t getNumChannels()
+  {
+    return getPublisherManifest().apply_info_size();
+  }
+  /**
+   * Adds to a supplied vector the set of channels this
+   * publisher publishes to.
+   *
+   * @param[out] Pointer to a Vector of channel IDs to add to
+   */
+  void getChannels(std::vector<uint32_t> *out_channels);
+private:
+  typedef drizzled::unordered_map<uint32_t, PublisherManifest::PublisherApplyInfo *> ChannelApplyMap;
+  static const std::string MANIFEST_FILE_NAME;
+  /** 
+   * Global publisher lock 
+   * @todo Break into a scoreboard?
+   */
+  pthread_mutex_t latch;
+  /**
+   * The manifest message which contains information about the
+   * latest transaction messages that were applied on the
+   * publishing node.  This is kept on disk and in-memory, but the
+   * information in it is only authoritative for the publisher's
+   * data state, not the subscribers.
+   */
+  PublisherManifest publisher_manifest;
+  /**
+   * Map of channel ID to the mutable publisher apply info
+   * for that channel.
+   */
+  ChannelApplyMap channel_apply_map;
+  /**
+   * The file for our manifest
+   */
+  int manifest_file;
+  /**
+   * Locks the publisher
+   */
+  void lock()
+  {
+    pthread_mutex_lock(&latch);
+  }
+  /**
+   * Unlocks the publisher
+   */
+  void unlock()
+  {
+    pthread_mutex_unlock(&latch);
+  }
+  /**
+   * Returns the publisher's manifest message
+   */
+  PublisherManifest &getPublisherManifest()
+  {
+    return publisher_manifest;
+  }
+  /**
+   * Reads the publisher's manifest file on startup
+   */
+  bool readManifestFile();
+  /**
+   * Writes the publisher's manifest file
+   */
+  bool writeManifestFile();
+  /**
+   * Syncs the publisher's manifest file
+   */
+  bool syncManifestFile();
+};
+
+#endif /* PLUGIN_TRANSACTION_LOG_PUBLISHER_H */

=== added file 'plugin/transaction_log/replication.proto'
--- plugin/transaction_log/replication.proto	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/replication.proto	2010-05-05 02:25:42 +0000
@@ -0,0 +1,59 @@
+/**
+ * A class representing this information
+ * a subscriber keeps about the data it
+ * has pulled from publishers
+ */
+message SubscriberManifest
+{
+  /**
+   * Sub message storing information about the
+   * communication this subsciber has had with
+   * a publisher.
+   */
+  message PublisherCommunicationInfo
+  {
+    required string hostname = 1; /* Possibly make this into a PublisherInfo class */
+    required uint32 channel_id = 2; /* Replication channel */
+    required uint64 last_communication_try = 3; /* Timestamp of last attempt at communication */
+    required uint64 last_communication_success = 4; /* Timestamp of last successful communication */
+    required uint64 last_received_transaction_id = 5; /* Transaction ID of last received transaction */
+    required uint64 last_received_end_timestamp = 6; /* End timestamp of last received transaction */
+    required uint64 last_applied_transaction_id = 7; /* Transaction ID of last applied transaction */
+    required uint64 last_applied_end_timestamp = 8; /* End timestamp of last applied transaction */
+  }
+  /* The set of information about communication with publishers */
+  repeated PublisherCommunicationInfo communication_info = 1;
+}
+
+message PublisherManifest
+{
+  /**
+   * Sub message storing information about the
+   * communication this subsciber has had with
+   * a publisher.
+   */
+  message PublisherApplyInfo
+  {
+    required uint32 channel_id = 1; /* Replication channel */
+    required uint64 last_applied_transaction_id = 2; /* ID of last applied transaction */
+    required uint64 last_applied_end_timestamp = 3; /* End timestamp of last applied transaction */
+  }
+  /**
+   * Sub message storing information about the
+   * communication this subsciber has had with
+   * a publisher.
+   */
+  message SubscriberCommunicationInfo
+  {
+    required string hostname = 1; /* Possibly make this into a SubscriberInfo class */
+    required uint32 channel_id = 2; /* Replication channel */
+    required uint64 last_communication_try = 3; /* Timestamp of last attempt at communication */
+    required uint64 last_communication_success = 4; /* Timestamp of last successful communication */
+    required uint64 last_sent_transaction_id = 5; /* Transaction ID of last sent transaction */
+    required uint64 last_sent_end_timestamp = 6; /* End timestamp of last snet transaction */
+  }
+  /* The set of information about channel applied transactions */
+  repeated PublisherApplyInfo apply_info = 1;
+  /* The set of information about communication with subscribers */
+  repeated SubscriberCommunicationInfo subscriber_info = 2;
+}

=== added directory 'plugin/transaction_log/tests/inc'
=== added file 'plugin/transaction_log/tests/inc/publisher.manifest'
--- plugin/transaction_log/tests/inc/publisher.manifest	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/inc/publisher.manifest	2010-05-05 02:25:42 +0000
@@ -0,0 +1,2 @@
+
+

¹è½¢Ã¥¡
\ No newline at end of file

=== modified file 'plugin/transaction_log/tests/r/bad_replicator.result'
--- plugin/transaction_log/tests/r/bad_replicator.result	2010-04-05 16:06:01 +0000
+++ plugin/transaction_log/tests/r/bad_replicator.result	2010-05-05 02:25:42 +0000
@@ -3,4 +3,5 @@
 WHERE DATA_DICTIONARY.PLUGINS.PLUGIN_NAME LIKE 'Transaction%';
 PLUGIN_NAME	PLUGIN_TYPE	IS_ACTIVE	MODULE_NAME
 transaction_log_applier	TransactionApplier	FALSE	transaction_log
+transaction_log_publisher	Publisher	TRUE	transaction_log
 SET GLOBAL transaction_log_truncate_debug= true;

=== added file 'plugin/transaction_log/tests/r/publisher_manifest.result'
--- plugin/transaction_log/tests/r/publisher_manifest.result	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/r/publisher_manifest.result	2010-05-05 02:25:42 +0000
@@ -0,0 +1,20 @@
+SET GLOBAL transaction_log_truncate_debug= true;
+DROP TABLE IF EXISTS t1;
+CREATE TABLE t1 (
+id INT NOT NULL PRIMARY KEY
+, padding VARCHAR(200) NOT NULL
+);
+INSERT INTO t1 VALUES (1, "I love testing.");
+INSERT INTO t1 VALUES (2, "I hate testing.");
+DROP TABLE t1;
+1var/master-data/publisher.manifest
+SET GLOBAL transaction_log_truncate_debug= true;
+testDefaultStartupVariables (__main__.TestPublisherManifest) ... ok
+testReadCorruptPublisherManifest (__main__.TestPublisherManifest) ... ok
+testReadPublisherManifest (__main__.TestPublisherManifest) ... ok
+testWritePublisherManifest (__main__.TestPublisherManifest) ... ok
+
+----------------------------------------------------------------------
+Ran 4 tests
+
+OK

=== added directory 'plugin/transaction_log/tests/replication'
=== added file 'plugin/transaction_log/tests/replication/COPYING'
--- plugin/transaction_log/tests/replication/COPYING	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/COPYING	2010-05-05 02:25:42 +0000
@@ -0,0 +1,32 @@
+Drizzle Replication Test Suite
+
+Copyright (C) 2010 Jay Pipes <[e-mail]>
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+    * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+
+    * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+
+    * The names of its contributors may not be used to endorse or
+promote products derived from this software without specific prior
+written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

=== added directory 'plugin/transaction_log/tests/replication/bootstrap'
=== added file 'plugin/transaction_log/tests/replication/bootstrap/ib_logfile0'
Binary files plugin/transaction_log/tests/replication/bootstrap/ib_logfile0	1970-01-01 00:00:00 +0000 and plugin/transaction_log/tests/replication/bootstrap/ib_logfile0	2010-05-05 02:25:42 +0000 differ
=== added file 'plugin/transaction_log/tests/replication/bootstrap/ib_logfile1'
Binary files plugin/transaction_log/tests/replication/bootstrap/ib_logfile1	1970-01-01 00:00:00 +0000 and plugin/transaction_log/tests/replication/bootstrap/ib_logfile1	2010-05-05 02:25:42 +0000 differ
=== added file 'plugin/transaction_log/tests/replication/bootstrap/ibdata1'
Binary files plugin/transaction_log/tests/replication/bootstrap/ibdata1	1970-01-01 00:00:00 +0000 and plugin/transaction_log/tests/replication/bootstrap/ibdata1	2010-05-05 02:25:42 +0000 differ
=== added directory 'plugin/transaction_log/tests/replication/inc'
=== added file 'plugin/transaction_log/tests/replication/inc/corrupted_publisher.manifest'
--- plugin/transaction_log/tests/replication/inc/corrupted_publisher.manifest	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/inc/corrupted_publisher.manifest	2010-05-05 02:25:42 +0000
@@ -0,0 +1,2 @@
+
+¹è½¢Ã¥¡

=== added file 'plugin/transaction_log/tests/replication/inc/publisher.manifest'
--- plugin/transaction_log/tests/replication/inc/publisher.manifest	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/inc/publisher.manifest	2010-05-05 02:25:42 +0000
@@ -0,0 +1,2 @@
+
+

2äò€˜Í§¡
\ No newline at end of file

=== added file 'plugin/transaction_log/tests/replication/test_publisher_manifest.py'
--- plugin/transaction_log/tests/replication/test_publisher_manifest.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/test_publisher_manifest.py	2010-05-05 02:25:42 +0000
@@ -0,0 +1,417 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.
+
+"""Drizzle Transaction Log Test Suite - Publisher Manifest
+
+This tests the startup process of the Drizzle server and
+the reading of the transaction log's publisher manifest
+"""
+
+__author__ = 'Jay Pipes <[e-mail]>'
+
+# 
+# This file contains a test for reading the transaction
+# log's publisher manifest file on startup.
+# 
+# It is necessary to use this testing suite
+# because the test-run.pl script assumes too many things
+# about the Drizzle test environment, and we need to 
+# initialize both proper and corrupted publisher manifests
+# in order to test the startup process correctly.
+#
+
+import os
+import sys
+import commands
+import unittest
+import shutil
+import re
+
+drizzled_root_dir= os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
+trx_log_root_dir= os.path.join(drizzled_root_dir, 'plugin', 'transaction_log')
+replication_test_root_dir= os.path.join(trx_log_root_dir, 'tests', 'replication')
+
+def compile_python_proto(protodir, protofile):
+  """Compiles a proto into the proto_pb2.py file and puts it into the main test directory"""
+  protocommand= "protoc --python_out=. --proto_path=%s %s" % (protodir, os.path.join(protodir, protofile))
+  (retcode, output)= commands.getstatusoutput(protocommand)
+  if retcode != 0:
+    raise Exception("Failed to compile Python protobuffers for %s\nTried command: %s" % (protofile, protocommand))
+
+compile_python_proto(trx_log_root_dir, 'replication.proto') # proto is in /plugin/transaction_log/
+
+import replication_pb2 as rpb
+
+from testlib.server import DrizzledServer
+from testlib.client import DrizzleClient
+
+class TestPublisherManifest(unittest.TestCase):
+
+  DEFAULT_START_OPTIONS= {
+    'mysql_protocol_port': DrizzledServer.DEFAULT_MYSQL_PROTOCOL_PORT,
+    'drizzle_protocol_port': DrizzledServer.DEFAULT_DRIZZLE_PROTOCOL_PORT,
+    'datadir': os.path.join(replication_test_root_dir, "var"),
+    'basedir': drizzled_root_dir
+  }
+
+  def setUp(self):
+    pass
+
+  def tearDown(self):
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.server.clear()
+    self.resetDataDir()
+
+  def resetDataDir(self):
+    """Copies default InnoDB data files from bootstrap into datadir to speed up starts"""
+    (retcode, ignored)= commands.getstatusoutput("rm -rf %s" % os.path.join(replication_test_root_dir, "var", "*"))
+
+  def bootstrapDataFiles(self):
+    """Copies default InnoDB data files from bootstrap into datadir to speed up starts"""
+    (retcode, ignored)= commands.getstatusoutput("cp -r %s %s" % (os.path.join(replication_test_root_dir, "bootstrap", "*"),
+                                                                  os.path.join(replication_test_root_dir, "var", "*")))
+
+  def testWritePublisherManifest(self):
+    #
+    # Here, we test the scenario of a newly
+    # started server with no data history.  There
+    # should be no publisher manifest file at startup
+    # and after plugin processing, the publisher manifest
+    # file should exist but have no size.  Once a few transactions
+    # are applied on the server, the manifest should show data
+    # fields with the last transaction ID and end timestamp
+    #
+    startup_options= {
+      'transaction-log-enable': '1'
+    }
+    for k, v in TestPublisherManifest.DEFAULT_START_OPTIONS.iteritems():
+      startup_options[k]= v
+
+    self.server= DrizzledServer(startup_options)
+
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.resetDataDir()
+    
+    self.bootstrapDataFiles()
+
+    # Test that the publisher manifest does not exist
+    # before startup.
+
+    manifest_filepath= os.path.join(replication_test_root_dir, "var", "publisher.manifest")
+
+    self.assertFalse(os.path.exists(manifest_filepath))
+
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+
+    # Test that a publisher manifest exists after startup
+    # but has no size since it has not been written to yet
+
+    self.assertTrue(os.path.exists(manifest_filepath))
+    self.assertEqual(os.path.getsize(manifest_filepath), 0)
+
+    # Now add a single transaction and verify the publisher
+    # manifest contains proper information by reading the
+    # manifest into a protobuffer container class object
+
+    client_options= {
+      'basedir': startup_options['basedir'],
+      'port': startup_options['mysql_protocol_port']
+    }
+    
+    self.client= DrizzleClient(client_options)
+    self.assertTrue(self.client.execute("CREATE SCHEMA test"))
+    self.assertTrue(self.client.execute("CREATE TABLE t1 (id INT NOT NULL PRIMARY KEY)", "test"))
+    self.assertTrue(self.client.execute("INSERT INTO t1 VALUES(1)", "test"))
+
+    # Create the manifest GPB class object
+    # by reading in the publisher manifest, and then
+    # check to ensure manifest data memebers are appropriate
+
+    f= open(manifest_filepath)
+
+    try:
+      manifest= rpb.PublisherManifest()
+      manifest.ParseFromString(f.read())
+    except Exception as e:
+      self.assertFalse(True)
+    
+    f.close()
+
+    self.assertEqual(len(manifest.apply_info), 1)
+    self.assertEqual(manifest.apply_info[0].channel_id, 1)
+    self.assertEqual(manifest.apply_info[0].last_applied_transaction_id, 3)
+    self.assertNotEqual(manifest.apply_info[0].last_applied_end_timestamp, 0)
+
+  def testReadCorruptPublisherManifest(self):
+    #
+    # Here, we test the scenario of a restarted server
+    # that has a corrupted publisher manifest.  We start
+    # the server after copying in a deliberately corrupted
+    # manifest file and check to ensure that the server
+    # appropriately disables the publisher and prints an
+    # error on startup.
+    #
+    startup_options= {
+      'transaction-log-enable': '1'
+    }
+    for k, v in TestPublisherManifest.DEFAULT_START_OPTIONS.iteritems():
+      startup_options[k]= v
+
+    self.server= DrizzledServer(startup_options)
+
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.resetDataDir()
+    
+    self.bootstrapDataFiles()
+
+    # Copy in the pre-corrupted manifest file into the datadir
+    manifest_filepath= os.path.join(replication_test_root_dir, "var", "publisher.manifest")
+    corrupted_manifest_filepath= os.path.join(replication_test_root_dir, "inc", "corrupted_publisher.manifest")
+    shutil.copy(corrupted_manifest_filepath, manifest_filepath)
+
+    # Test that a publisher manifest exists after startup
+    # but that the publisher has been disabled due to a corrupt
+    # publisher manifest and the server is prevented from
+    # starting up
+
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+
+    f= open(os.path.join(replication_test_root_dir, "var", "error.log"))
+    error_log_text= f.read()
+
+    # grep the error log and ensure that an error message was
+    # thrown by the publisher constructor
+
+    self.assertTrue(error_log_text.find("corrupted data") > 0)
+
+    f.close()
+
+    # Check that the publisher plugin is not available
+    
+    client_options= {
+      'basedir': startup_options['basedir'],
+      'port': startup_options['mysql_protocol_port']
+    }
+    self.client= DrizzleClient(client_options)
+
+    expected_results_1= [
+      {
+      'COUNT(*)': '0'
+      }
+    ]
+    results= self.client.fetchAllAssoc("SELECT COUNT(*) FROM DATA_DICTIONARY.REPLICATION_PUBLISHERS")
+    
+    self.assertTrue(len(results) == len(expected_results_1))
+    
+    rowno= 0
+    for row in results:
+      for k, v in row.iteritems():
+        self.assertEqual(str(expected_results_1[rowno][k]), str(v))
+      rowno+= 1
+
+  def testReadPublisherManifest(self):
+    #
+    # Here, we test the scenario of a restarted server
+    # that has a stored publisher manifest.  We start
+    # the server after copying in a deliberately constructed 
+    # manifest file and check to ensure that the server
+    # appropriately reads the manifest and updates its state
+    #
+    startup_options= {
+      'transaction-log-enable': '1'
+    }
+    for k, v in TestPublisherManifest.DEFAULT_START_OPTIONS.iteritems():
+      startup_options[k]= v
+
+    self.server= DrizzledServer(startup_options)
+
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.resetDataDir()
+    
+    self.bootstrapDataFiles()
+
+    # Copy in the pre-corrupted manifest file into the datadir
+    manifest_filepath= os.path.join(replication_test_root_dir, "var", "publisher.manifest")
+    corrupted_manifest_filepath= os.path.join(replication_test_root_dir, "inc", "publisher.manifest")
+    shutil.copy(corrupted_manifest_filepath, manifest_filepath)
+
+    # Test that a publisher manifest exists after startup
+    # but that the publisher has been disabled due to a corrupt
+    # publisher manifest and the server is prevented from
+    # starting up
+
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+
+    f= open(os.path.join(replication_test_root_dir, "var", "error.log"))
+    error_log_text= f.read()
+
+    # grep the error log and ensure that an error message was
+    # thrown by the publisher constructor
+
+    self.assertEqual(error_log_text.find("corrupted data"), -1)
+
+    f.close()
+
+    # Check that the publisher plugin is available
+    
+    client_options= {
+      'basedir': startup_options['basedir'],
+      'port': startup_options['mysql_protocol_port']
+    }
+    self.client= DrizzleClient(client_options)
+
+    expected_results_1= [
+      {
+      'COUNT(*)': '1'
+      }
+    ]
+    results= self.client.fetchAllAssoc("SELECT COUNT(*) FROM DATA_DICTIONARY.REPLICATION_PUBLISHERS")
+    
+    self.assertTrue(len(results) == len(expected_results_1))
+    
+    rowno= 0
+    for row in results:
+      for k, v in row.iteritems():
+        self.assertEqual(str(expected_results_1[rowno][k]), str(v))
+      rowno+= 1
+
+    # Check that the publisher reports a transaction ID of 50
+    # and an end timestamp of 1272396191381860, which is what the pre-constructed
+    # manifest file contains
+
+    expected_results_2= [
+      {
+      'PUBLISHER': 'transaction_log_publisher',
+      'CHANNEL_ID': 1,
+      'LAST_APPLIED_TRANSACTION_ID': 50,
+      'LAST_APPLIED_END_TIMESTAMP': 1272396191381860 
+      }
+    ]
+    results= self.client.fetchAllAssoc("SELECT * FROM DATA_DICTIONARY.REPLICATION_PUBLISHER_CHANNELS")
+    
+    self.assertTrue(len(results) == len(expected_results_2))
+    
+    rowno= 0
+    for row in results:
+      for k, v in row.iteritems():
+        self.assertEqual(str(expected_results_2[rowno][k]), str(v))
+      rowno+= 1
+
+  def testDefaultStartupVariables(self):
+    #
+    # Here, we test the "default" scenario of a newly
+    # started server with no data history. We check to see
+    # if the replication_publisher and replication_publisher_channels
+    # views in the data_dictionary contain proper values 
+    #
+    startup_options= {
+      'transaction-log-enable': '1'
+    }
+    for k, v in TestPublisherManifest.DEFAULT_START_OPTIONS.iteritems():
+      startup_options[k]= v
+      
+    client_options= {
+      'basedir': startup_options['basedir'],
+      'port': startup_options['mysql_protocol_port']
+    }
+
+    self.server= DrizzledServer(startup_options)
+
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.resetDataDir()
+    
+    self.bootstrapDataFiles()
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+    self.client= DrizzleClient(client_options)
+
+    expected_results_1= [
+      {
+      'PUBLISHER_NAME': 'transaction_log_publisher'
+      }
+    ]
+    results= self.client.fetchAllAssoc("SELECT * FROM DATA_DICTIONARY.REPLICATION_PUBLISHERS")
+    
+    self.assertTrue(len(results) == len(expected_results_1))
+    
+    rowno= 0
+    for row in results:
+      for k, v in row.iteritems():
+        self.assertEqual(str(expected_results_1[rowno][k]), str(v))
+      rowno+= 1
+
+    expected_results_2= [
+      {
+      'PUBLISHER': 'transaction_log_publisher',
+      'CHANNEL_ID': 1,
+      'LAST_APPLIED_TRANSACTION_ID': 0,
+      'LAST_APPLIED_END_TIMESTAMP': 0
+      }
+    ]
+    results= self.client.fetchAllAssoc("SELECT * FROM DATA_DICTIONARY.REPLICATION_PUBLISHER_CHANNELS")
+    
+    self.assertTrue(len(results) == len(expected_results_2))
+    
+    rowno= 0
+    for row in results:
+      for k, v in row.iteritems():
+        self.assertEqual(str(expected_results_2[rowno][k]), str(v))
+      rowno+= 1
+
+#
+# Code below taken verbatim from Eric Day's prototest
+# test suite in the mysql_protocol Drizzle plugin
+#
+import optparse
+
+parser = optparse.OptionParser()
+
+parser.add_option("-t", "--test", dest="testcase", default=None,
+                  help="Test case to run", metavar="TESTCASE")
+
+(options, args) = parser.parse_args()
+
+class CustomTestRunner(unittest.TextTestRunner):
+  def run(self, test):
+    result = self._makeResult()
+    test(result)
+    result.printErrors()
+    self.stream.writeln(result.separator2)
+    run = result.testsRun
+    self.stream.writeln("Ran %d test%s" % (run, run != 1 and "s" or ""))
+    self.stream.writeln()
+    if not result.wasSuccessful():
+      self.stream.write("FAILED (")
+      failed, errored = map(len, (result.failures, result.errors))
+      if failed:
+        self.stream.write("failures=%d" % failed)
+      if errored:
+        if failed: self.stream.write(", ")
+        self.stream.write("errors=%d" % errored)
+      self.stream.writeln(")")
+    else:
+      self.stream.writeln("OK")
+    return result
+
+if __name__ == '__main__':
+  if options.testcase is None:
+    suite = unittest.TestLoader().loadTestsFromModule(__import__('__main__'))
+  else:
+    suite = unittest.TestLoader().loadTestsFromTestCase(eval(options.testcase))
+  CustomTestRunner(stream=sys.stdout, verbosity=2).run(suite)

=== added directory 'plugin/transaction_log/tests/replication/testlib'
=== added file 'plugin/transaction_log/tests/replication/testlib/__init__.py'
--- plugin/transaction_log/tests/replication/testlib/__init__.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/testlib/__init__.py	2010-05-05 02:25:42 +0000
@@ -0,0 +1,10 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.

=== added file 'plugin/transaction_log/tests/replication/testlib/client.py'
--- plugin/transaction_log/tests/replication/testlib/client.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/testlib/client.py	2010-05-05 02:25:42 +0000
@@ -0,0 +1,137 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.
+
+"""Defines the adapter for controlling a Drizzle database client"""
+
+import os
+import commands
+import sys
+import time
+import unittest
+
+class DrizzleClientException(Exception):
+  pass
+
+class DrizzleClient:
+  """A class wrapping a Drizzled client"""
+
+  DEFAULT_PORT= 9306
+
+  def __init__(self, options= {}):
+    self.start_options= options
+    self.cleanStartOptions()
+
+  def getPort(self):
+    """Returns port the client connects to"""
+    return self.start_options["port"]
+
+  def cleanStartOptions(self):
+    """Cleans up the startup options dictionary"""
+    clean_start_options= {}
+    for key in self.start_options.keys():
+      # Ensure option names are output as --option_name
+      # and clean the self.start_options array to remove any leading --
+      # and convert - to _
+      value= self.start_options[key]
+      
+      if not key.startswith("--"):
+        clean_key= key.replace('-','_')
+      else:
+        clean_key= key[2:].replace("-","_")
+      clean_start_options[clean_key]= value
+    
+    self.start_options= clean_start_options
+
+  def getStartOptionString(self):
+    """Builds the string of options to pass to drizzled on startup"""
+    out_options= ["%s=%s" % ("--" + key, value) for key, value in self.start_options.iteritems()]
+
+    return " ".join(out_options)
+
+  def execute(self, statement, schema=None):
+    """Executes a supplied SQL statement"""
+    if schema is None:
+      schema_string= ""
+    else:
+      schema_string= "--database=%s" % schema
+    client_cmd= "%s --port=%d %s -e\"%s\"" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , schema_string
+                                        , statement)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL:\n\"%s\"\nGot error: %s" % (self.start_options["port"], statement, output)
+      return False
+    return True
+
+  def executeFile(self, file):
+    """Executes SQL statements in a supplied file"""
+    client_cmd= "%s --port=%d < %s" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , file)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL in file:\n\"%s\"\nGot error: %s" % (self.start_options["port"], file, output)
+      return False
+    return True
+
+  def fetchAsDict(self, statement):
+    """Executes a supplied SELECT or SHOW statement and returns a dictionary with first column as keys, second column as values."""
+    client_cmd= "%s --port=%d -e\"%s\"" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , statement)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL:\n\"%s\"\nGot error: %s" % (self.start_options["port"], statement, output)
+      return {}
+    else:
+      # Now we build the results dictionary...
+      results= {}
+      # The first line is the field names separated by a tab
+      # Every line after is a tab-delimited field values.
+      lines= output.split("\n")
+      fields= lines[0].split("\t")
+      for line in lines[1:]:
+        data_fields= line.split("\t")
+        results[data_fields[0]]= data_fields[1]
+      return results
+
+  def fetchAllAssoc(self, statement):
+    """Executes a supplied SELECT or SHOW statement and returns a sequence with associative dictionary of results."""
+    client_cmd= "%s --port=%d -e\"%s\"" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , statement)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL:\n\"%s\"\nGot error: %s" % (self.start_options["port"], statement, output)
+      return []
+    else:
+      # Now we build the results dictionary...
+      results= []
+      # The first line is the field names separated by a tab
+      # Every line after is a tab-delimited field values.
+      lines= output.split("\n")
+      fields= lines[0].split("\t")
+      for line in lines[1:]:
+        row= {}
+        data_fields= line.split("\t")
+        fieldno= 0
+        for field in fields:
+          row[field]= data_fields[fieldno]
+          fieldno= fieldno + 1
+        results.append(row)
+      return results
+
+class TestDrizzleClient(unittest.TestCase):
+  pass
+
+if __name__ == '__main__':
+  unittest.main()

=== added file 'plugin/transaction_log/tests/replication/testlib/server.py'
--- plugin/transaction_log/tests/replication/testlib/server.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/testlib/server.py	2010-05-05 02:25:42 +0000
@@ -0,0 +1,152 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.
+
+"""Defines the adapter for controlling a Drizzle database server"""
+
+import os
+import commands
+import sys
+import time
+import unittest
+
+class DrizzledServerStartupException(Exception):
+  pass
+
+class DrizzledServerShutdownException(Exception):
+  pass
+
+class DrizzledServer:
+  """A class responsible for starting, stopping, and pinging a Drizzled server"""
+
+  DEFAULT_MYSQL_PROTOCOL_PORT= 9306
+  DEFAULT_DRIZZLE_PROTOCOL_PORT= 9307
+
+  def __init__(self, options= {}):
+    self.start_options= options
+    self.defaults_file= ""
+    self.cleanStartOptions()
+
+  def getPort(self, protocol="drizzle"):
+    """Returns port the server listens for supplied protocol"""
+    return self.start_options[protocol + "_protocol_port"]
+
+  def cleanStartOptions(self):
+    """Cleans up the startup options dictionary"""
+    clean_start_options= {}
+    for key in self.start_options.keys():
+      # Ensure option names are output as --option_name
+      # and clean the self.start_options array to remove any leading --
+      # and convert - to _
+      value= self.start_options[key]
+      
+      if not key.startswith("--"):
+        clean_key= key.replace('-','_')
+      else:
+        clean_key= key[2:].replace("-","_")
+
+      # The defaults-file must ALWAYS be the first command-line
+      # option given to a drizzled server, so we strip it and
+      # store it locally here...
+      if clean_key == 'defaults_file':
+        self.defaults_file= "--defaults-file=%s" % value
+      else:
+        clean_start_options[clean_key]= value
+    self.start_options= clean_start_options
+
+  def getStartOptionString(self):
+    """Builds the string of options to pass to drizzled on startup"""
+    out_options= ["%s=%s" % ("--" + key, value) for key, value in self.start_options.iteritems()]
+
+    return " ".join(out_options)
+
+  def start(self):
+    """Starts a drizzled server"""
+
+    for x in ("mysql_protocol_port","basedir","datadir"):
+      if x not in self.start_options.keys():
+        raise DrizzledServerStartupException("Attempt to start server without required startup options")
+
+    start_option_string= self.getStartOptionString()
+    start_cmd= "%s %s %s > %s 2>&1 &" % (os.path.join(self.start_options['basedir'], "drizzled", "drizzled"),
+                                         self.defaults_file,
+                                         start_option_string,
+                                         os.path.join(self.start_options['datadir'], 'error.log'))
+    server_output= os.system(start_cmd)
+
+    # Here, we sleep until the server is up and running or until a timeout occurs...
+    timeout= 10
+    timer= 0
+    while not self.ping() and timer != timeout:
+      time.sleep(.25)
+      timer= timer + 1
+
+    return True
+
+  def ping(self, quiet= False):
+    """Pings the server. Returns True if server is up and running, False otherwise."""
+    ping_cmd= "%s --ping --port=%d" % (os.path.join(self.start_options['basedir'], "client", "drizzle"),
+                                       self.start_options['mysql_protocol_port'])
+
+    (retcode, output)= commands.getstatusoutput(ping_cmd)
+
+    return retcode == 0
+
+  def clear(self):
+    """Clears data files for the server."""
+    (retcode, ignored)= commands.getstatusoutput("rm -rf %s; mkdir %s" % (self.start_options['datadir'], self.start_options['datadir']))
+
+  def stop(self):
+    """Stops the server"""
+    shutdown_cmd= "%s --shutdown --port=%d" % (os.path.join(self.start_options['basedir'], "client", "drizzle"),
+                                               self.start_options['mysql_protocol_port'])
+
+    (retcode, output)= commands.getstatusoutput(shutdown_cmd)
+
+    return retcode == 0
+
+  def restart(self):
+    self.stop()
+    return self.start()
+
+  def stopAll(self):
+    """Stops ALL drizzled servers"""
+    (retcode, num_drizzled_running)= commands.getstatusoutput("ps aux | grep drizzled | wc -l")
+    if retcode == 0 and int(num_drizzled_running) > 0:
+      (retcode, ignored)= commands.getstatusoutput("killall -9 drizzled")
+      time.sleep(3)
+
+class TestDrizzledServer(unittest.TestCase):
+
+  DEFAULT_START_OPTIONS= {
+    'mysql_protocol_port': DrizzledServer.DEFAULT_MYSQL_PROTOCOL_PORT,
+    'drizzle_protocol_port': DrizzledServer.DEFAULT_DRIZZLE_PROTOCOL_PORT,
+    'datadir': os.path.join(os.getcwd().partition("plugin")[0], "tests", "var"),
+    'basedir': os.path.join(os.getcwd().partition("plugin")[0], "drizzled")
+  }
+
+  def setUp(self):
+    self.server= DrizzledServer(TestDrizzledServer.DEFAULT_START_OPTIONS)
+    if self.server.ping():
+      self.server.stopAll()
+
+  def tearDown(self):
+    self.server.stop()
+                    
+  def testStartAndPing(self):
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+                    
+  def testStartAndStop(self):
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.stop())
+
+if __name__ == '__main__':
+  unittest.main()

=== added directory 'plugin/transaction_log/tests/replication/var'
=== added file 'plugin/transaction_log/tests/t/default-master.opt'
--- plugin/transaction_log/tests/t/default-master.opt	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/t/default-master.opt	2010-05-05 02:25:42 +0000
@@ -0,0 +1,1 @@
+--transaction-log-enable --scheduler=multi_thread

=== added symlink 'plugin/transaction_log/tests/t/publisher_manifest-master.opt'
=== target is u'default-master.opt'
=== added file 'plugin/transaction_log/tests/t/publisher_manifest.test'
--- plugin/transaction_log/tests/t/publisher_manifest.test	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/t/publisher_manifest.test	2010-05-05 02:25:42 +0000
@@ -0,0 +1,19 @@
+#
+# Tests that the publisher manifest file is written
+# at shutdown
+#
+
+# Truncate the log file to reset for the test
+--source ../plugin/transaction_log/tests/t/truncate_log.inc
+
+# Populate log with some records...
+--source ../plugin/transaction_log/tests/t/insert.inc
+
+# Trim result since Solaris/BSD wc program apparently adds whitespace before output
+--exec wc -l var/master-data/publisher.manifest | tr -d ' ' 
+
+# Truncate the log file to reset for the next test
+--source ../plugin/transaction_log/tests/t/truncate_log.inc
+
+# Execute the external test suite for replication
+--exec $DRIZZLE_TEST_DIR/../plugin/transaction_log/tests/replication/test_publisher_manifest.py

=== modified file 'plugin/transaction_log/transaction_log.cc'
--- plugin/transaction_log/transaction_log.cc	2010-03-31 19:04:12 +0000
+++ plugin/transaction_log/transaction_log.cc	2010-05-05 02:25:42 +0000
@@ -291,7 +291,7 @@
   drizzled::TransactionServices::singleton().resetTransactionId();
 }
 
-bool TransactionLog::findLogFilenameContainingTransactionId(const ReplicationServices::GlobalTransactionId&,
+bool TransactionLog::findLogFilenameContainingTransactionId(const TransactionServices::TransactionId&,
                                                             string &out_filename) const
 {
   /* 

=== modified file 'plugin/transaction_log/transaction_log.h'
--- plugin/transaction_log/transaction_log.h	2010-03-27 18:51:59 +0000
+++ plugin/transaction_log/transaction_log.h	2010-05-05 02:25:42 +0000
@@ -39,7 +39,7 @@
 #define PLUGIN_TRANSACTION_LOG_TRANSACTION_LOG_H
 
 #include <drizzled/atomics.h>
-#include <drizzled/replication_services.h>
+#include <drizzled/transaction_services.h>
 
 #include "transaction_log_entry.h"
 
@@ -162,7 +162,7 @@
    * @retval
    *  false otherwise
    */
-  bool findLogFilenameContainingTransactionId(const drizzled::ReplicationServices::GlobalTransactionId &to_find,
+  bool findLogFilenameContainingTransactionId(const drizzled::TransactionServices::TransactionId &to_find,
                                               std::string &out_filename) const;
 
   /**

=== modified file 'plugin/transaction_log/transaction_log_applier.cc'
--- plugin/transaction_log/transaction_log_applier.cc	2010-04-12 19:28:43 +0000
+++ plugin/transaction_log/transaction_log_applier.cc	2010-05-05 02:25:42 +0000
@@ -47,6 +47,7 @@
 #include "transaction_log.h"
 #include "transaction_log_applier.h"
 #include "transaction_log_index.h"
+#include "publisher.h"
 
 #include <vector>
 
@@ -62,10 +63,12 @@
 TransactionLogApplier::TransactionLogApplier(const string name_arg,
                                              TransactionLog *in_transaction_log,
                                              TransactionLogIndex *in_transaction_log_index,
+                                             TransactionLogPublisher *in_transaction_log_publisher,
                                              uint32_t in_num_write_buffers) :
   plugin::TransactionApplier(name_arg),
   transaction_log(in_transaction_log),
   transaction_log_index(in_transaction_log_index),
+  transaction_log_publisher(in_transaction_log_publisher),
   num_write_buffers(in_num_write_buffers),
   write_buffers()
 {
@@ -119,5 +122,7 @@
                                                       entry_size),
                                   to_apply,
                                   checksum);
+  /* Tell the publisher about our apply */
+  transaction_log_publisher->setLastApplied(to_apply);
   return plugin::SUCCESS;
 }

=== modified file 'plugin/transaction_log/transaction_log_applier.h'
--- plugin/transaction_log/transaction_log_applier.h	2010-04-12 19:15:00 +0000
+++ plugin/transaction_log/transaction_log_applier.h	2010-05-05 02:25:42 +0000
@@ -50,6 +50,7 @@
 
 class TransactionLog;
 class TransactionLogIndex;
+class TransactionLogPublisher;
 class WriteBuffer;
 
 class TransactionLogApplier: public drizzled::plugin::TransactionApplier 
@@ -58,6 +59,7 @@
   TransactionLogApplier(const std::string name_arg,
                         TransactionLog *in_transaction_log,
                         TransactionLogIndex *in_transaction_log_index,
+                        TransactionLogPublisher *in_transaction_log_publisher,
                         uint32_t in_num_write_buffers);
 
   /** Destructor */
@@ -89,10 +91,11 @@
   TransactionLogApplier &operator=(const TransactionLogApplier &other);
   /** 
    * This Applier owns the memory of the associated TransactionLog 
-   * and its index - so we have to track it. 
+   * and its index and publisher - so we have to track it. 
    */
   TransactionLog *transaction_log;
   TransactionLogIndex *transaction_log_index;
+  TransactionLogPublisher *transaction_log_publisher;
   uint32_t num_write_buffers; ///< Number of write buffers used
   std::vector<WriteBuffer *> write_buffers; ///< array of write buffers
 

=== modified file 'plugin/transaction_log/transaction_log_reader.cc'
--- plugin/transaction_log/transaction_log_reader.cc	2010-04-23 01:02:40 +0000
+++ plugin/transaction_log/transaction_log_reader.cc	2010-05-05 02:25:42 +0000
@@ -65,7 +65,7 @@
 using namespace drizzled;
 using namespace google;
 
-bool TransactionLogReader::read(const ReplicationServices::GlobalTransactionId &to_read_trx_id, 
+bool TransactionLogReader::read(const TransactionServices::TransactionId &to_read_trx_id, 
                             message::Transaction *to_fill)
 {
   /* 

=== modified file 'plugin/transaction_log/transaction_log_reader.h'
--- plugin/transaction_log/transaction_log_reader.h	2010-03-05 18:08:49 +0000
+++ plugin/transaction_log/transaction_log_reader.h	2010-05-05 02:25:42 +0000
@@ -69,7 +69,7 @@
    * @retval
    *  false if not found or read successfully
    */
-  bool read(const drizzled::ReplicationServices::GlobalTransactionId &to_read_trx_id, 
+  bool read(const drizzled::TransactionServices::TransactionId &to_read_trx_id, 
             drizzled::message::Transaction *to_fill);
 };
 

=== modified file 'tests/r/data_dictionary_like_info.result'
--- tests/r/data_dictionary_like_info.result	2010-04-07 21:54:30 +0000
+++ tests/r/data_dictionary_like_info.result	2010-05-05 02:25:42 +0000
@@ -191,7 +191,7 @@
  or DATA_TYPE = 'varchar')
 group by DATA_TYPE order by DATA_TYPE, num;
 DATA_TYPE	group_concat(table_schema, '.', table_name)	num
-VARCHAR	DATA_DICTIONARY.CHARACTER_SETS,DATA_DICTIONARY.CHARACTER_SETS,DATA_DICTIONARY.CHARACTER_SETS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.GLOBAL_STATEMENTS,DATA_DICTIONARY.GLOBAL_STATEMENTS,DATA_DICTIONARY.GLOBAL_STATUS,DATA_DICTIONARY.GLOBAL_STATUS,DATA_DICTIONARY.GLOBAL_VARIABLES,DATA_DICTIONARY.GLOBAL_VARIABLES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEX_PART	145
+VARCHAR	DATA_DICTIONARY.CHARACTER_SETS,DATA_DICTIONARY.CHARACTER_SETS,DATA_DICTIONARY.CHARACTER_SETS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLLATIONS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.COLUMNS,DATA_DICTIONARY.GLOBAL_STATEMENTS,DATA_DICTIONARY.GLOBAL_STATEMENTS,DATA_DICTIONARY.GLOBAL_STATUS,DATA_DICTIONARY.GLOBAL_STATUS,DATA_DICTIONARY.GLOBAL_VARIABLES,DATA_DICTIONARY.GLOBAL_VARIABLES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEXES,DATA_DICTIONARY.INDEX_PART	147
 Warnings:
 Warning	1260	1 line(s) were cut by GROUP_CONCAT()
 create table t1(f1 char(1) not null, f2 char(9) not null);

=== modified file 'tests/r/information_schema.result'
Binary files tests/r/information_schema.result	2010-04-05 16:30:06 +0000 and tests/r/information_schema.result	2010-05-05 02:25:42 +0000 differ