From d304a4b2b19dca115954f209d6a37a6958e73ab9 Mon Sep 17 00:00:00 2001
From: Boris Kolpackov <boris@codesynthesis.com>
Date: Fri, 18 Jan 2013 10:23:39 +0200
Subject: Add support for post-commit/rollback callbacks

New test: common/transaction/callback.
---
 common/session/custom/driver.cxx  | 201 ++++++++++++++++++++++----------------
 common/session/custom/session.cxx |  28 +++++-
 common/session/custom/session.hxx |  27 +++--
 common/session/custom/session.txx |  11 ++-
 4 files changed, 168 insertions(+), 99 deletions(-)

(limited to 'common/session')

diff --git a/common/session/custom/driver.cxx b/common/session/custom/driver.cxx
index 0a58ea5..1b00ada 100644
--- a/common/session/custom/driver.cxx
+++ b/common/session/custom/driver.cxx
@@ -30,16 +30,14 @@ using odb::transaction;
 struct counting_tracer: odb::tracer
 {
   virtual void
-  execute (odb::connection&, const char*)
-  {
-    count++;
-  }
-
+  execute (odb::connection&, const char*) {count++;}
   size_t count;
 };
 
 static counting_tracer tracer;
 
+struct failed {};
+
 int
 main (int argc, char* argv[])
 {
@@ -81,99 +79,138 @@ main (int argc, char* argv[])
       t.commit ();
     }
 
-    session s;
-    shared_ptr<employer> st, cs;
-    shared_ptr<employee> ste, cse;
-
     {
-      transaction t (db->begin ());
+      session s;
+      shared_ptr<employer> st, cs;
+      shared_ptr<employee> ste, cse;
 
-      st = db->load<employer> ("Simple Tech Ltd");
-      ste = db->load<employee> (st->employees ()[0].object_id ());
+      {
+        transaction t (db->begin ());
 
-      // Test object cache.
-      //
-      shared_ptr<employee> e (st->employees ()[0].load ());
-      assert (ste->employer () == st);
-      assert (ste == e);
+        st = db->load<employer> ("Simple Tech Ltd");
+        ste = db->load<employee> (st->employees ()[0].object_id ());
 
-      t.commit ();
-    }
+        // Test object cache.
+        //
+        shared_ptr<employee> e (st->employees ()[0].load ());
+        assert (ste->employer () == st);
+        assert (ste == e);
 
-    {
-      transaction t (db->begin ());
+        t.commit ();
+      }
 
-      cs = db->load<employer> ("Complex Systems Inc");
-      cse = db->load<employee> (cs->employees ()[0].object_id ());
-      cs->employees ()[0].load ();
+      {
+        transaction t (db->begin ());
 
-      t.commit ();
-    }
+        cs = db->load<employer> ("Complex Systems Inc");
+        cse = db->load<employee> (cs->employees ()[0].object_id ());
+        cs->employees ()[0].load ();
 
-    cs->symbol ("CSI");
-
-    // Swap employees.
-    //
-    ste->employer (cs);
-    cse->employer (st);
-    st->employees ()[0] = cse;
-    cs->employees ()[0] = ste;
+        t.commit ();
+      }
 
-    {
-      transaction t (db->begin ());
-      tracer.count = 0;
-      t.tracer (tracer);
-      s.flush (*db); // Flush all the changes.
-      assert (tracer.count == 3);
-      t.commit ();
-      s.mark (); // Mark all the changed objects as unchanged.
-    }
+      cs->symbol ("CSI");
 
-    {
-      transaction t (db->begin ());
-      tracer.count = 0;
-      t.tracer (tracer);
-      s.flush (*db);
-      assert (tracer.count == 0);
-      t.commit ();
-    }
-
-    cs->symbol ("COMP");
-    st->symbol ("SMPL");
-
-    {
-      transaction t (db->begin ());
-      tracer.count = 0;
-      t.tracer (tracer);
-      s.flush (*db);
-      assert (tracer.count == 2);
-      t.commit ();
-      s.mark ();
-    }
-
-    {
-      transaction t (db->begin ());
-      tracer.count = 0;
-      t.tracer (tracer);
-      s.flush (*db);
-      assert (tracer.count == 0);
-      t.commit ();
+      // Swap employees.
+      //
+      ste->employer (cs);
+      cse->employer (st);
+      st->employees ()[0] = cse;
+      cs->employees ()[0] = ste;
+
+      {
+        transaction t (db->begin ());
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 3);
+        t.commit ();
+      }
+
+      {
+        transaction t (db->begin ());
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 0);
+        t.commit ();
+      }
+
+      cs->symbol ("COMP");
+      st->symbol ("SMPL");
+
+      {
+        transaction t (db->begin ());
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 2);
+        t.commit ();
+      }
+
+      {
+        transaction t (db->begin ());
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 0);
+        t.commit ();
+      }
+
+      // Explicit update.
+      //
+      cs->symbol ("CS");
+      st->symbol ("ST");
+
+      {
+        transaction t (db->begin ());
+        db->update (cs);
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 1);
+        t.commit ();
+      }
+
+      // Rollback after update.
+      //
+      cs->symbol ("CSI");
+
+      try
+      {
+        transaction t (db->begin ());
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 1);
+        throw failed ();
+        t.commit ();
+      }
+      catch (const failed&)
+      {
+        transaction t (db->begin ());
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 1);
+        t.commit ();
+      }
     }
 
-    // Explicit update.
+    // Test session destruction before transaction is commited.
     //
-    cs->symbol ("CS");
-    st->symbol ("ST");
-
     {
       transaction t (db->begin ());
-      db->update (cs);
-      tracer.count = 0;
-      t.tracer (tracer);
-      s.flush (*db);
-      assert (tracer.count == 1);
+      {
+        session s;
+        shared_ptr<employer> st (db->load<employer> ("Simple Tech Ltd"));
+        st->symbol ("STL");
+        tracer.count = 0;
+        t.tracer (tracer);
+        s.flush (*db);
+        assert (tracer.count == 1);
+      }
       t.commit ();
-      s.mark ();
     }
   }
   catch (const odb::exception& e)
diff --git a/common/session/custom/session.cxx b/common/session/custom/session.cxx
index f8cd102..0009d79 100644
--- a/common/session/custom/session.cxx
+++ b/common/session/custom/session.cxx
@@ -10,6 +10,7 @@ session* session::current;
 
 session::
 session ()
+    : tran_ (0)
 {
   assert (current == 0);
   current = this;
@@ -18,6 +19,11 @@ session ()
 session::
 ~session ()
 {
+  // Unregister from transaction.
+  //
+  if (tran_ != 0)
+    tran_->unregister (this);
+
   assert (current == this);
   current = 0;
 }
@@ -25,13 +31,27 @@ session::
 void session::
 flush (odb::database& db)
 {
+  bool flushed (false);
+
   for (type_map::iterator i (map_.begin ()), e (map_.end ()); i != e; ++i)
-    i->second->flush (db);
+  {
+    bool r (i->second->flush (db));
+    flushed = flushed || r;
+  }
+
+  // If we flushed anything, then register the post-commit/rollback callback.
+  //
+  if (flushed)
+  {
+    tran_ = &odb::transaction::current ();
+    tran_->register_ (&mark, this, odb::transaction::event_all, 0, &tran_);
+  }
 }
 
 void session::
-mark ()
+mark (unsigned short event, void* key, unsigned long long)
 {
-  for (type_map::iterator i (map_.begin ()), e (map_.end ()); i != e; ++i)
-    i->second->mark ();
+  session& s (*static_cast<session*> (key));
+  for (type_map::iterator i (s.map_.begin ()), e (s.map_.end ()); i != e; ++i)
+    i->second->mark (event);
 }
diff --git a/common/session/custom/session.hxx b/common/session/custom/session.hxx
index 3b1789a..ce7e43c 100644
--- a/common/session/custom/session.hxx
+++ b/common/session/custom/session.hxx
@@ -10,6 +10,8 @@
 #include <typeinfo>
 
 #include <odb/database.hxx>
+#include <odb/transaction.hxx>
+
 #include <odb/traits.hxx>            // odb::object_traits
 #include <odb/details/type-info.hxx> // odb::details::type_info_comparator
 
@@ -36,28 +38,26 @@ public:
 
   // Change tracking interface.
   //
+public:
   // Call flush() within a transaction to apply the changes to the
-  // database. Then after successfully committing the transaction,
-  // call mark() to mark all the changed objects as again unchanged.
+  // database.
   //
-public:
   void
   flush (odb::database&);
 
-  void
-  mark ();
-
 private:
   struct object_map_base
   {
     virtual
     ~object_map_base () {}
 
-    virtual void
+    // Return true we flushed anything.
+    //
+    virtual bool
     flush (odb::database&) = 0;
 
     virtual void
-    mark () = 0;
+    mark (unsigned short event) = 0;
   };
 
   enum object_state
@@ -85,11 +85,11 @@ private:
                      std::map<typename odb::object_traits<T>::id_type,
                               object_data<T> >
   {
-    virtual void
+    virtual bool
     flush (odb::database&);
 
     virtual void
-    mark ();
+    mark (unsigned short event);
   };
 
   // Object cache interface.
@@ -150,10 +150,17 @@ public:
   erase (odb::database&, const typename odb::object_traits<T>::id_type&);
 
 private:
+  // Post-commit/rollback callback.
+  //
+  static void
+  mark (unsigned short event, void* key, unsigned long long);
+
+private:
   typedef std::map<const std::type_info*,
                    std::shared_ptr<object_map_base>,
                    odb::details::type_info_comparator> type_map;
   type_map map_;
+  odb::transaction* tran_;
 };
 
 #include "session.txx"
diff --git a/common/session/custom/session.txx b/common/session/custom/session.txx
index fd90b6a..31c08c7 100644
--- a/common/session/custom/session.txx
+++ b/common/session/custom/session.txx
@@ -128,9 +128,10 @@ erase (odb::database&, const typename odb::object_traits<T>::id_type& id)
 }
 
 template <typename T>
-void session::object_map<T>::
+bool session::object_map<T>::
 flush (odb::database& db)
 {
+  bool r (false);
   for (typename object_map<T>::iterator i (this->begin ()), e (this->end ());
        i != e; ++i)
   {
@@ -138,12 +139,16 @@ flush (odb::database& db)
 
     if (d.state == changed || d.obj->changed (*d.orig))
       db.update (d.obj); // State changed by the update() notification.
+
+    r = r || d.state == flushed;
   }
+
+  return r;
 }
 
 template <typename T>
 void session::object_map<T>::
-mark ()
+mark (unsigned short event)
 {
   for (typename object_map<T>::iterator i (this->begin ()), e (this->end ());
        i != e; ++i)
@@ -151,6 +156,6 @@ mark ()
     object_data<T>& d (i->second);
 
     if (d.state == flushed)
-      d.state = tracking;
+      d.state = event == odb::transaction::event_commit ? tracking : changed;
   }
 }
-- 
cgit v1.1