THRIFT-1021. ocaml: Framed transport support for OCaml

Patch: Richard Low

git-svn-id: https://svn.apache.org/repos/asf/thrift/trunk@1063917 13f79535-47bb-0310-9956-ffa450edef68

diff --git a/lib/ocaml/src/Makefile b/lib/ocaml/src/Makefile
index 42ec8db..a97ade5 100644 (file)
--- a/lib/ocaml/src/Makefile
+++ b/lib/ocaml/src/Makefile
@@ -17,7 +17,7 @@
 # under the License.
 #
 
-SOURCES = Thrift.ml TBinaryProtocol.ml TSocket.ml TChannelTransport.ml TServer.ml TSimpleServer.ml TServerSocket.ml TThreadedServer.ml
+SOURCES = Thrift.ml TBinaryProtocol.ml TSocket.ml TFramedTransport.ml TChannelTransport.ml TServer.ml TSimpleServer.ml TServerSocket.ml TThreadedServer.ml
 RESULT = thrift
 LIBS = unix threads
 THREADS = yes

diff --git a/lib/ocaml/src/TFramedTransport.ml b/lib/ocaml/src/TFramedTransport.ml
new file mode 100644 (file)
index 0000000..1be51e7
--- /dev/null
+++ b/lib/ocaml/src/TFramedTransport.ml
@@ -0,0 +1,93 @@
+open Thrift
+
+module T = Transport
+
+let c_0xff_32 = Int32.of_string "0xff"
+
+(* Copied from OCamlnet rtypes.ml *)
+let encode_frame_size x =
+       let s = String.create 4 in
+       let n3 = Int32.to_int (Int32.shift_right_logical x 24) land 0xff in
+       let n2 = Int32.to_int (Int32.shift_right_logical x 16) land 0xff in
+       let n1 = Int32.to_int (Int32.shift_right_logical x 8) land 0xff in
+       let n0 = Int32.to_int (Int32.logand x c_0xff_32) in
+               String.unsafe_set s 0 (Char.unsafe_chr n3);
+               String.unsafe_set s 1 (Char.unsafe_chr n2);
+               String.unsafe_set s 2 (Char.unsafe_chr n1);
+               String.unsafe_set s 3 (Char.unsafe_chr n0);
+               s
+               
+let decode_frame_size s = 
+       let n3 = Int32.of_int (Char.code s.[0]) in
+       let n2 = Int32.of_int (Char.code s.[1]) in
+       let n1 = Int32.of_int (Char.code s.[2]) in
+       let n0 = Int32.of_int (Char.code s.[3]) in
+               Int32.logor
+               (Int32.shift_left n3 24)
+               (Int32.logor
+                       (Int32.shift_left n2 16)
+                       (Int32.logor
+                               (Int32.shift_left n1 8)
+                               n0))
+
+class t ?(max_length=Sys.max_string_length) (transport: T.t) =
+object (self)
+       inherit T.t
+
+       method isOpen = transport#isOpen
+       method opn = transport#opn
+       method close = transport#close
+ 
+       val mutable read_buf = None
+       val mutable read_buf_offset = 0
+       val mutable write_buf = ""
+
+       method private read_frame =
+               let len_buf = String.create 4 in
+               assert (transport#readAll len_buf 0 4 = 4); 
+               
+               let size = Int32.to_int (decode_frame_size len_buf) in
+               
+               (if size < 0
+               then failwith (Printf.sprintf "Read a negative frame size (%i)!" size));
+               
+               (if size > max_length
+               then failwith (Printf.sprintf "Frame size (%i) larger than max length (%i)!" size max_length));
+
+               let buf = String.create size in
+                       assert (transport#readAll buf 0 size = size);
+                       read_buf <- Some buf;
+                       read_buf_offset <- 0
+
+       method private read_from_frame frame buf off len =
+               let to_copy = min len ((String.length frame) - read_buf_offset) in
+                       String.blit frame read_buf_offset buf off to_copy;
+                       read_buf_offset <- read_buf_offset + to_copy;
+                       to_copy
+
+       method read buf off len = 
+               match read_buf with
+               | Some frame -> 
+                       let i = self#read_from_frame frame buf off len in
+                       if i > 0
+                       then i
+                       else begin
+                               self#read_frame; 
+                               self#read_from_frame frame buf off len
+                       end
+               | None ->
+                               self#read_frame;
+                               self#read buf off len 
+        
+       method write buf off len = 
+               write_buf <- write_buf ^ (String.sub buf off len)
+
+       method flush = 
+               let encoded_size = encode_frame_size (Int32.of_int (String.length write_buf)) in
+                       transport#write encoded_size 0 (String.length encoded_size);
+                       transport#write write_buf 0 (String.length write_buf);
+                       transport#flush; 
+                       write_buf <- ""
+end
+
+