Support stored procedure.

lib/Connection.cpp

@@ -71,6 +71,26 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 //#define PRINTMARK() fprintf(stderr, "%08x:%s:%s MARK(%d)\n", GetTickCount(), __FILE__, __FUNCTION__, __LINE__)		
 #define PRINTMARK() 		
 
+#ifdef DEBUG
+static void hexdump(void *ptr, int buflen) {
+  unsigned char *buf = (unsigned char*)ptr;
+  int i, j;
+  for (i=0; i<buflen; i+=16) {
+    printf("%06x: ", i);
+    for (j=0; j<16; j++) 
+      if (i+j < buflen)
+        printf("%02x ", buf[i+j]);
+      else
+        printf("   ");
+    printf(" ");
+    for (j=0; j<16; j++) 
+      if (i+j < buflen)
+        printf("%c", isprint(buf[i+j]) ? buf[i+j] : '.');
+    printf("\n");
+  }
+}
+#endif
+
 Connection::Connection (UMConnectionCAPI *_capi) 
   :	m_reader(MYSQL_RX_BUFFER_SIZE)
   , m_writer(MYSQL_TX_BUFFER_SIZE)
@@ -84,6 +104,7 @@ Connection::Connection (UMConnectionCAPI *_capi)
   memcpy (&m_capi, _capi, sizeof (UMConnectionCAPI));
   m_dbgMethodProgress = 0;
   m_errorType = UME_OTHER;
+  m_has_more_result = false;
 }
 
 Connection::~Connection()
@@ -163,6 +184,11 @@ bool Connection::readSocket()
         return false;
       }
 
+#ifdef DEBUG
+    printf("recv %d\n", recvResult);
+    hexdump(m_reader.getWritePtr(), recvResult);
+#endif
+
       m_reader.push (recvResult);
 
       return true;
@@ -188,6 +214,11 @@ bool Connection::writeSocket()
       return false;
     }
 
+#ifdef DEBUG
+    printf("send %d\n", sendResult);
+    hexdump(m_writer.getReadCursor(), sendResult);
+#endif
+
     m_writer.pull(sendResult);
     return true;
 }
@@ -288,6 +319,9 @@ bool Connection::processHandshake()
     m_clientCaps  &= ~MCP_NO_SCHEMA;
     m_clientCaps &= ~MCP_SSL;
 
+    if(serverVersion[0]=='5')
+        m_clientCaps |= MCP_MULTI_RESULTS;
+
     if (!(serverCaps & MCP_CONNECT_WITH_DB) && !m_database.empty())
     {
       setError("Protocol < 4.1 not supported", 3, UME_OTHER);
@@ -571,6 +605,7 @@ void *Connection::handleOKPacket()
 
   m_reader.skip();
 
+  m_has_more_result = serverStatus & SERVER_MORE_RESULTS_EXISTS;
   return m_capi.resultOK(affectedRows, insertId, serverStatus, (char *) message, len);
 }
 
@@ -675,6 +710,12 @@ void *Connection::handleResultPacket(int _fieldCount)
 
     if (result == 0xfe)
     {
+      // ignore warning count.
+      m_reader.readBytes(2); 
+
+      UINT16 serverStatus = m_reader.readShort();
+      m_has_more_result = serverStatus & SERVER_MORE_RESULTS_EXISTS;
+
       m_reader.skip();
       break;
     }
@@ -710,6 +751,7 @@ void *Connection::handleResultPacket(int _fieldCount)
 void *Connection::query(const char *_query, size_t _cbQuery)
 {
   m_dbgMethodProgress ++;
+  m_has_more_result = false;
 
   if (m_dbgMethodProgress > 1)
   {
@@ -752,6 +794,13 @@ void *Connection::query(const char *_query, size_t _cbQuery)
     return NULL;
   }
 
+  return nextResultSet();
+}
+
+
+void *Connection::nextResultSet()
+{
+
   if (!recvPacket())
   {
     PRINTMARK();

lib/Connection.h

@@ -99,6 +99,7 @@ class Connection
   PacketWriter m_writer;
   UINT32 m_clientCaps;
   std::string m_query;
+  bool m_has_more_result;
 
   std::string m_errorMessage;
   int m_errno;
@@ -118,6 +119,8 @@ class Connection
   bool connect(const char *_host, int _port, const char *_username, const char *_password, const char *_database, int *_autoCommit, MYSQL_CHARSETS _charset);
   //void handleSocketEvent (SocketEvents _evt);
   void *query(const char *_query, size_t _cbQuery);
+  bool hasMoreResult() { return m_has_more_result; }
+  void *nextResultSet();
   bool getLastError (const char **_ppOutMessage, int *_outErrno, int *_outErrorType);
 
   int getRxBufferSize();
@@ -145,4 +148,4 @@ class Connection
 protected:
 };
 
-#endif
+#endif

lib/capi.cpp

@@ -79,6 +79,16 @@ EXPORT_ATTR void * UMConnection_Query(UMConnection conn, const char *_query, siz
   return ((Connection *)conn)->query(_query, _cbQuery);
 }
 
+EXPORT_ATTR int UMConnection_HasMoreResult(UMConnection conn)
+{
+  return (int)((Connection *)conn)->hasMoreResult();
+}
+
+EXPORT_ATTR void * UMConnection_NextResultSet(UMConnection conn)
+{
+  return ((Connection *)conn)->nextResultSet();
+}
+
 EXPORT_ATTR int UMConnection_Connect (UMConnection conn, const char *_host, int _port, const char *_username, const char *_password, const char *_database, int *_autoCommit, int _charset)
 {
   return ((Connection *)conn)->connect(_host, _port, _username, _password, _database, _autoCommit, (MYSQL_CHARSETS) _charset) ? 1 : 0;

lib/mysqldefs.h

@@ -377,6 +377,20 @@ enum MYSQL_FIELDFLAG
   MFFLAG_SET_FLAG = 0x0800,
 };
 
+enum SERVER_STATUS
+{
+  SERVER_STATUS_IN_TRANS =              0x0001,
+  SERVER_STATUS_AUTOCOMMIT =            0x0002,
+  SERVER_MORE_RESULTS_EXISTS =          0x0008,
+  SERVER_QUERY_NO_GOOD_INDEX_USED =     0x0010,
+  SERVER_QUERY_NO_INDEX_USED =          0x0020,
+  SERVER_STATUS_CURSOR_EXISTS =         0x0040,
+  SERVER_STATUS_LAST_ROW_SENT =         0x0080,
+  SERVER_STATUS_DB_DROPPED =            0x0100,
+  SERVER_STATUS_NO_BACKSLASH_ESCAPES =  0x0200,
+  SERVER_STATUS_METADATA_CHANGED =      0x0400,
+};
+
 #define MYSQL_PACKET_HEADER_SIZE 4
 #define MYSQL_PROTOCOL_VERSION 0x0a
 #define MYSQL_PACKET_SIZE (1024 * 1024 * 16)

python/umysql.c

@@ -1236,6 +1236,26 @@ PyObject *Connection_query(Connection *self, PyObject *args)
 
 
 
+PyObject *Connection_nextset(Connection *self, PyObject *notused)
+{
+  void *ret;
+
+  if (!UMConnection_HasMoreResult(self->conn))
+  {
+    Py_RETURN_NONE;
+  }
+
+  ret = UMConnection_NextResultSet(self->conn);
+  if (ret == NULL)
+  {
+      return HandleError(self, "nextset");
+  }
+
+  PRINTMARK();
+  return (PyObject *) ret;
+}
+
+
 
 PyObject *Connection_close(Connection *self, PyObject *notused)
 {
@@ -1256,6 +1276,7 @@ static void Connection_Destructor(Connection *self)
 static PyMethodDef Connection_methods[] = {
   {"connect", (PyCFunction)			Connection_connect,			METH_VARARGS, "Connects to database server. Arguments: host, port, username, password, database, autocommit, charset"},
   {"query", (PyCFunction)				Connection_query,				METH_VARARGS, "Performs a query. Arguments: query, arguments to escape"},
+  {"nextset", (PyCFunction)				Connection_nextset,				METH_NOARGS, "Try to fetch the next result set, if there no more sets, return None."},
   {"close", (PyCFunction)	Connection_close,	METH_NOARGS, "Closes connection"},
   {"is_connected", (PyCFunction) Connection_isConnected, METH_NOARGS, "Check connection status"},
   {"settimeout", (PyCFunction) Connection_setTimeout, METH_VARARGS, "Sets connection timeout in seconds"},

python/umysql.h

@@ -119,6 +119,8 @@ typedef void * UMConnection;
 UMConnection UMConnection_Create(UMConnectionCAPI *_capi);
 void UMConnection_Destroy(UMConnection _conn);
 void *UMConnection_Query(UMConnection conn, const char *_query, size_t _cbQuery);
+int UMConnection_HasMoreResult(UMConnection conn);
+void *UMConnection_NextResultSet(UMConnection conn);
 int  UMConnection_Connect (UMConnection conn, const char *_host, int _port, const char *_username, const char *_password, const char *_database, int *_autoCommit, int _charset);
 int UMConnection_GetLastError (UMConnection conn, const char **_ppOutMessage, int *_outErrno, int *_type);
 int UMConnection_GetTxBufferSize (UMConnection conn);

tests/testdb.sql

@@ -18,4 +18,24 @@ CREATE TABLE `tblautoincbigint` (
   PRIMARY KEY(test_id)  
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1;
 
+DELIMITER //
+CREATE PROCEDURE CreateTest(IN id int(11), IN str varchar(1024))
+BEGIN
+  INSERT INTO tbltest VALUES (id, str, NULL);
+END //
+
+CREATE PROCEDURE QueryTest(IN str varchar(1024), OUT rowcount int(11))
+BEGIN
+  SELECT COUNT(*) INTO rowcount FROM tbltest WHERE test_string=str;
+  SELECT * FROM tbltest WHERE test_string=str;
+END //
+
+CREATE PROCEDURE TestMultiResult()
+BEGIN
+    SELECT 1;
+    SELECT 2;
+    SELECT 3;
+END //
+DELIMITER ;
+
 GRANT ALL on gevent_test.* to 'gevent_test'@'localhost' identified by 'gevent_test';

tests/tests.py

@@ -760,6 +760,24 @@ def testPercentEscaping(self):
 
         cnn.close()
 
+    def testProcedure(self):
+        cnn = umysql.Connection()
+        cnn.connect (DB_HOST, 3306, DB_USER, DB_PASSWD, DB_DB)
+        cnn.query("truncate tbltest")
+        self.assertEquals((1, 0), cnn.query('call CreateTest(%s, %s)', (1, 'test')))
+        rs = cnn.query('call QueryTest(%s, @count)', ('test', ))
+        self.assertEquals([(1, 'test', None)], rs.rows)
+        self.assertEquals((0, 0), cnn.query('select @count'))
+
+    def testMultiResult(self):
+        cnn = umysql.Connection()
+        cnn.connect (DB_HOST, 3306, DB_USER, DB_PASSWD, DB_DB)
+        self.assertEquals([(1,)], cnn.query("call TestMultiResult()").rows)
+        self.assertEquals([(2,)], cnn.nextset().rows)
+        self.assertEquals([(3,)], cnn.nextset().rows)
+        self.assertEquals((0,0), cnn.nextset())
+        self.assertEquals(None, cnn.nextset())
+
 if __name__ == '__main__':
     from guppy import hpy
     hp = hpy()