Implement LOB read and write support using OCI callbacks

3 files changed, 196 insertions, 26 deletions

diff --git a/odb/oracle/oracle-types.hxx b/odb/oracle/oracle-types.hxx
index 4cdd7bb..1f5fe57 100644
--- a/odb/oracle/oracle-types.hxx
+++ b/odb/oracle/oracle-types.hxx
@@ -17,13 +17,57 @@ namespace odb
   {
     struct bind
     {
-      ub2 type;          // The type stored by buffer. This must be an
-                         // external OCI type identifier of the form SQLT_XXX.
+      ub2 type;       // The type stored by buffer. This must be an external
+                      // OCI type identifier of the form SQLT_XXX.
       void* buffer;
-      ub2* size;         // The number of bytes in buffer.
-      sb4 capacity;      // The maximum number of bytes that can be stored in
-                         // buffer.
-      sb2* indicator;    // Pointer to an OCI indicator variable.
+      ub2* size;      // The number of bytes in buffer. Due to
+                      // inconsistencies in the OCI interface, this will be
+                      // interpreted as a ub4 when callbacks are in use.
+      sb4 capacity;   // The maximum number of bytes that can be stored in
+                      // buffer.
+      sb2* indicator; // Pointer to an OCI indicator variable.
+
+      // Callback function signature used to specify parameters that are
+      // passed to the database.
+      //
+      typedef bool (*param_callback_type) (
+        void** context,    // [in/out] The user context.
+        void** buffer,     // [out] A a buffer containing the parameter data
+                           // to write to the database.
+        ub4* size,         // [out] The parameter data length in bytes.
+        bool* null,        // [out] True if the parameter is null. Both buffer
+                           // and size are ignored is this is true.
+        bool* last,        // [out] True if buffer contains the last piece of
+                           // data.
+        void* temp_buffer, // [in] A temporary buffer that may be used if
+                           // required. The 'buffer' parameter should specify
+                           // this buffer on return if it is used.
+        ub4 capacity);     // [in] The temporary buffer length in bytes.
+
+      // Callback function signature used to specify values returned from the
+      // database.
+      //
+      typedef bool (*result_callback_type) (
+        void** context, // [in/out] The user context.
+        void* buffer,   // [in] A buffer containing the result data.
+        ub4 size,       // [in] The result data length in bytes.
+        bool null,      // [in] True if the result data is NULL.
+        bool last);     // [in] True if this is the last piece of result data.
+
+      param_callback_type param_callback;
+      result_callback_type result_callback;
+
+      // This pointer is provided to the user through the 'context' parameter
+      // in both parameter and result callback functions.
+      //
+      void* callback_context;
+
+      // This flag is used exclusively during parameter callback invocation.
+      // If set, it indicates that this is the first time the callback is
+      // being invoked for this bind instance, and thus OCI is requesting
+      // the first piece of parameter data.
+      //
+      bool first_piece;
     };
   }
 }
diff --git a/odb/oracle/statement.cxx b/odb/oracle/statement.cxx
index 5aa025e..447ef06 100644
--- a/odb/oracle/statement.cxx
+++ b/odb/oracle/statement.cxx
@@ -3,6 +3,8 @@
 // copyright : Copyright (c) 2005-2011 Code Synthesis Tools CC
 // license   : ODB NCUEL; see accompanying LICENSE file
 
+#include <cassert>
+
 #include <oci.h>
 
 #include <odb/exceptions.hxx> // object_not_persistent
@@ -17,6 +19,76 @@ namespace odb
 {
   namespace oracle
   {
+    static sb4
+    param_callback_proxy (void* context,
+                          OCIBind*,
+                          ub4,            // iteration
+                          ub4,            // index
+                          void** buffer,
+                          ub4* len,
+                          ub1* piece,
+                          void** indicator)
+    {
+      bind& b (*static_cast<bind*> (context));
+
+      bool last (false), null (false);
+
+      if (!(*b.param_callback) (&b.callback_context,
+                                buffer,
+                                len,
+                                &null,
+                                &last,
+                                b.buffer,
+                                b.capacity))
+        return OCI_ROWCBK_DONE;
+
+
+      *b.indicator = null ? -1 : 0;
+      *indicator = &b.indicator;
+
+      if (null)
+        *piece = OCI_ONE_PIECE;
+      else if (b.first_piece)
+      {
+        b.first_piece = false;
+        *piece = last ? OCI_ONE_PIECE : OCI_FIRST_PIECE;
+      }
+      else if (last)
+        *piece = OCI_LAST_PIECE;
+      else
+        *piece = OCI_NEXT_PIECE;
+
+      return OCI_CONTINUE;
+    }
+
+    static sb4
+    result_callback_proxy (void* context,
+                           OCIDefine*,
+                           ub4,                // iteration
+                           void** buffer,
+                           ub4** len,
+                           ub1* piece,
+                           void** indicator,
+                           ub2**)              // return_code)
+    {
+      bind& b (*static_cast<bind*> (context));
+
+      if (*piece == OCI_NEXT_PIECE)
+        if (!(*b.result_callback) (&b.callback_context,
+                                   &b.buffer,
+                                   *reinterpret_cast<ub4*> (b.size),
+                                   false,
+                                   false))
+          return OCI_ROWCBK_DONE;
+
+      *buffer = b.buffer;
+      *len = reinterpret_cast<ub4*> (b.size);
+      **len = static_cast<ub4> (b.capacity);
+      *indicator = &b.indicator;
+
+      return OCI_CONTINUE;
+    }
+
     //
     // statement
     //
@@ -47,50 +119,94 @@ namespace odb
     void statement::
     bind_param (bind* b, size_t c, size_t o)
     {
-      for (size_t i (0); i < c; ++i)
+      OCIError* err (conn_.error_handle ());
+
+      // The parameter position in OCIBindByPos is specified as a 1-based
+      // index. Convert 'o' to a 1-based offset.
+      //
+      ++o;
+
+      for (size_t e (o + c); o < e; ++c, ++b)
       {
         OCIBind* h (0);
 
         sword r (OCIBindByPos (stmt_,
                                &h,
-                               conn_.error_handle (),
-                               o + i,
-                               b[i].buffer,
-                               b[i].capacity,
-                               b[i].type,
-                               b[i].indicator,
-                               b[i].size,
+                               err,
+                               o,
+                               b->buffer,
+                               b->capacity,
+                               b->type,
+                               b->indicator,
+                               b->size,
                                0,
                                0,
                                0,
-                               OCI_DEFAULT));
+                               b->param_callback != 0 ?
+                               OCI_DATA_AT_EXEC : OCI_DEFAULT));
 
         if (r == OCI_ERROR || r == OCI_INVALID_HANDLE)
-          translate_error (conn_.error_handle (), r);
+          translate_error (err, r);
+
+        if (b->param_callback != 0)
+        {
+          r = OCIBindDynamic (h, err, b, &param_callback_proxy, 0, 0);
+
+          if (r == OCI_ERROR || r == OCI_INVALID_HANDLE)
+            translate_error (err, r);
+        }
       }
     }
 
     void statement::
     bind_result (bind* b, size_t c)
     {
-      for (size_t i (0); i < c; ++i)
+      OCIError* err (conn_.error_handle ());
+
+      // The parameter position in OCIDefineByPos is specified as a 1-based
+      // index.
+      //
+      for (size_t i (1); i <= c; ++i, ++b)
       {
         OCIDefine* h (0);
 
         sword r (OCIDefineByPos (stmt_,
                                  &h,
-                                 conn_.error_handle (),
+                                 err,
                                  i,
-                                 b[i].buffer,
-                                 b[i].capacity,
-                                 b[i].type,
-                                 b[i].indicator,
-                                 b[i].size,
+                                 b->buffer,
+                                 b->capacity,
+                                 b->type,
+                                 b->indicator,
+                                 b->size,
                                  0,
-                                 OCI_DEFAULT));
+                                 b->result_callback != 0 ?
+                                 OCI_DYNAMIC_FETCH : OCI_DEFAULT));
 
         if (r == OCI_ERROR || r == OCI_INVALID_HANDLE)
-          translate_error (conn_.error_handle (), r);
+          translate_error (err, r);
+
+        if (b->result_callback != 0)
+        {
+          r = OCIDefineDynamic (h, err, b, &result_callback_proxy);
+
+          if (r == OCI_ERROR || r == OCI_INVALID_HANDLE)
+            translate_error (err, r);
+        }
+      }
+    }
+
+    void statement::
+    finalize_result (bind* b, std::size_t c)
+    {
+      for (size_t i (0); i < c; ++i)
+      {
+        if (b[i].result_callback != 0)
+          (*b[i].result_callback) (&b[i].callback_context,
+                                   b[i].buffer,
+                                   *reinterpret_cast<ub4*> (b[i].size),
+                                   *b[i].indicator == -1,
+                                   true);
       }
     }
 
@@ -114,6 +230,7 @@ namespace odb
                       binding& cond,
                       binding& data)
         : statement (conn, s),
+          data_ (data),
           done_ (false)
     {
       bind_param (cond.bind, cond.count, 0);
@@ -128,7 +245,8 @@ namespace odb
 
       // @@ Retrieve a single row into the already bound output buffers
       // as an optimization? This will avoid multiple server round-trips
-      // in the case of a single object load.
+      // in the case of a single object load. Remember to invoke
+      // finalize_result on a successful prefetch.
       //
       sword r (OCIStmtExecute (conn_.handle (),
                                stmt_,
@@ -178,6 +296,8 @@ namespace odb
           translate_error (conn_.error_handle (), r);
         else if (r == OCI_NO_DATA)
           done_ = true;
+        else
+          finalize_result (data_.bind, data_.count);
       }
 
       return done_ ? no_data : success;
diff --git a/odb/oracle/statement.hxx b/odb/oracle/statement.hxx
index e999134..ffdb91a 100644
--- a/odb/oracle/statement.hxx
+++ b/odb/oracle/statement.hxx
@@ -50,6 +50,11 @@ namespace odb
       void
       bind_result (bind*, std::size_t count);
 
+      // Finalize the result by invoking the user provided callbacks for the
+      // final time.
+      //
+      void finalize_result (bind*, std::size_t count);
+
     protected:
       connection& conn_;
       auto_handle<OCIStmt> stmt_;
@@ -85,6 +90,7 @@ namespace odb
       select_statement& operator= (const select_statement&);
 
     private:
+      binding& data_;
       bool done_;
     };