Automatic type detection in extended query protocol

example/main.pony

@@ -1,7 +1,7 @@
 """
-pg.pony
+main.pony
+
 
-Do pg stuff.
 """
 use "pg"
 use "pg/codec"
@@ -11,6 +11,28 @@ use "debug"
 actor Main
   let sess: Session
 
+  new create(env: Env) =>
+
+    sess = Session(env where user="macflytest",
+                   password=EnvPasswordProvider(env),
+                   database="macflytest")
+
+    let that = recover tag this end
+    sess.raw("SELECT 42, 24;;",
+             recover val
+              lambda(r: Rows val)(that) =>
+                  that.raw_handler(r)
+              end
+             end)
+
+    sess.execute("SELECT $1, $2",
+                 recover val [as PGValue: I32(70000), I32(-100000)] end,
+                 recover val
+                  lambda(r: Rows val)(that) =>
+                      that.execute_handler(r)
+                  end
+                end)
+
   be raw_handler(rows: Rows val) =>
     for d in rows.desc.fields.values() do
       Debug.out(d.name)
@@ -21,23 +43,14 @@ actor Main
       end
     end
 
-  new create(env: Env) =>
-    sess = Session(env where user="macflytest", password=EnvPasswordProvider(env), database="macflytest")
-    let that = recover tag this end
-    sess.raw("SELECT 42, 24;;", recover val lambda(r: Rows val)(that) => that.raw_handler(r) end end)
-    sess.execute("SELECT $1, $2", recover val [as PGValue: I32(70000), I32(-100000)] end, recover val
-      lambda(rows: Rows val)(env, sess) =>
-        for d in rows.desc.fields.values() do
-          Debug.out(d.type_oid)
-          Debug.out(d.name)
-        end
-        Debug.out(rows.size())
-        for row in rows.values() do
-          for value in row.values() do
-            try Debug.out(value as I32) else Debug.out("error...") end
-          end
-        end
-        sess.terminate()
+  be execute_handler(rows: Rows val) =>
+    for d in rows.desc.fields.values() do
+      Debug.out(d.type_oid)
+      Debug.out(d.name)
+    end
+    for row in rows.values() do
+      for value in row.values() do
+        try Debug.out(value as I32) else Debug.out("error...") end
       end
-    end)
-
+    end
+    sess.terminate()

pg/codec/registry.pony

@@ -3,6 +3,43 @@ use "debug"
 
 use "pg"
 
+primitive TypeOid
+  """ The type oids are found with:
+
+    SELECT
+        oid,
+        typname
+    FROM                  
+        pg_catalog.pg_type
+    WHERE                    
+        typtype IN ('b', 'p')                                                 
+        AND (typelem = 0 OR typname = '_oid' OR typname='_text' OR typlen > 0)
+        AND oid <= 9999
+    ORDER BY
+        oid;
+
+  """
+  // TODO: Find NULL oid, i'm pretty sure it's not 0
+  fun apply(t: None): I32 => 0
+  fun apply(t: Bool val): I32 => 16
+  fun apply(t: U8 val): I32 => 18
+  fun apply(t: I64 val): I32 => 20
+  fun apply(t: I16 val): I32 => 21
+  fun apply(t: I32 val): I32 => 23
+  fun apply(t: String val): I32 => 25
+  fun apply(t: F32 val): I32 => 700
+  fun apply(t: F64 val): I32 => 701
+  /*fun apply(t: Any val): I32 => 0*/
+
+primitive TypeOids
+  fun apply(t: Array[PGValue] val): Array[I32] val =>
+    recover val
+      let result = Array[I32](t.size())
+      for item in t.values() do
+        try result.push(TypeOid(item) as I32) end
+      end
+      result
+    end
 
 primitive Decode
   fun apply(type_oid: I32, value: Array[U8] val, 0): PGValue ? =>
@@ -14,7 +51,7 @@ primitive Decode
 primitive DecodeText
   fun apply(23, value: Array[U8] val): I32 ? =>
     String.from_array(value).i32()
-  fun apply(type_oid: I32, value: Array[U8] val) ? => error
+  fun apply(type_oid: I32, value: Array[U8] val) ? => Debug.out("Unknown type OID: " + type_oid.string()); error
 
 primitive DecodeBinary
   fun apply(23, value: Array[U8] val): I32 ? => 
@@ -24,7 +61,7 @@ primitive DecodeBinary
     end
     result
 
-  fun apply(type_oid: I32, value: Array[U8] val) ? => error
+  fun apply(type_oid: I32, value: Array[U8] val) ? => Debug.out("Unknown type OID: " + type_oid.string()); error
 
 primitive EncodeBinary
   fun apply(param: I32, writer: Writer) ? =>

pg/connection/conversation.pony

@@ -96,7 +96,7 @@ actor ExecuteConversation is Conversation
     _conn.writev(recover val FlushMessage.done() end)
 
   be apply(c: BEConnection tag) =>
-    c.writev(recover val ParseMessage(query, "", recover [as I32: 23, 23] end).done() end)
+    c.writev(recover val ParseMessage(query, "", TypeOids(params)).done() end)
     _flush()
 
   be _bind() =>

pg/connection/tcp.pony

@@ -29,22 +29,22 @@ class PGNotify is TCPConnectionNotify
     _conn.connected()
 
   fun ref received(conn: TCPConnection ref, data: Array[U8] iso) =>
-    Debug.out("received")
+    // Debug.out("received")
     _listener.received(consume data)
 
   fun ref closed(conn: TCPConnection ref) =>
     /*_listener.received(recover [as U8: 0, 0, 0, 0, 0]end)*/
     _listener.terminate()
     _conn.received(ConnectionClosedMessage)
 
-  fun ref sent(conn: TCPConnection ref, data: (String val | Array[U8 val] val)): (String val | Array[U8 val] val) =>
-    Debug.out("send")
-    match data
-    | let s: String val => for c in s.values() do Debug.out(c) end
-    | let s: Array[U8 val] val => for c in s.values() do Debug.out(c) end
-    end
-    conn.write_final(data)
-    ""
+  /*fun ref sent(conn: TCPConnection ref, data: (String val | Array[U8 val] val)): (String val | Array[U8 val] val) =>*/
+    /*Debug.out("send")*/
+    /*match data*/
+    /*| let s: String val => for c in s.values() do Debug.out(c) end*/
+    /*| let s: Array[U8 val] val => for c in s.values() do Debug.out(c) end*/
+    /*end*/
+    /*conn.write_final(data)*/
+    /*""*/
 
 actor _Connection is BEConnection
   let _conn: TCPConnection tag
@@ -110,7 +110,7 @@ actor _Connection is BEConnection
     None
 
   be received(s: ServerMessage val) =>
-    Debug.out("recieved " + s.string())
+    // Debug.out("recieved " + s.string())
     _current.message(s)
 
   be _log_error(m: ErrorMessage val) =>

pg/pg.pony

@@ -42,7 +42,6 @@ class PGValueIterator is Iterator[Array[PGValue val]]
       let fmt = _desc.fields(idx).format
       idx = idx + 1
       result.push(Decode(typ, value.data, fmt))
-      Debug.out("###")
     end
     result
 

pg/protocol/client.pony

@@ -126,7 +126,7 @@ class CloseMessage is ClientMessage
 class ParseMessage is ClientMessage
   let _base: ClientMessageBase delegate ClientMessage = ClientMessageBase
 
-  new create(query: String, name: String, param_types: Array[I32]) =>
+  new create(query: String, name: String, param_types: Array[I32] val) =>
     _write(name)
     _zero()
     _write(query)