0
@@ -12,6 +12,7 @@
0
 
0
 #ifndef HAVE_STRLCPY
0
 
0
+static
0
 size_t
0
 strlcpy (char *dst, const char *src, size_t dst_sz)
0
 {
0
@@ -392,7 +393,6 @@
0
   name_buf[0] = '@';
0
   strlcpy(&name_buf[1], spec->name, sizeof(name_buf) - 1);
0
   
0
-  // TODO(kevinclark): Replace with the strlcpy method used in read
0
   VALUE value = rb_ivar_get(obj, rb_intern(name_buf));
0
   
0
   if (Qnil == value) {
0
0
@@ -500,13 +500,13 @@
0
 } field_header;
0
 
0
 typedef struct {
0
-  int key_type;
0
-  int val_type;
0
+  int8_t key_type;
0
+  int8_t val_type;
0
   int num_entries;
0
 } map_header;
0
 
0
 typedef struct {
0
-  intt type;
0
+  int8_t type;
0
   int num_elements;
0
 } list_header;
0
 
0
0
0
0
0
0
0
0
0
@@ -527,28 +527,89 @@
0
 #define read_struct_begin(buf)
0
 #define read_struct_end(buf)
0
 
0
-static void consume(decode_buffer* buf, int32_t size) {
0
+// This prototype is required to be able to run a call through rb_protect
0
+// which rescues from ruby exceptions
0
+static VALUE protectable_consume(VALUE args) {
0
+  VALUE trans = rb_ary_entry(args, 0);
0
+  VALUE size = rb_ary_entry(args, 1);
0
+  
0
+  return rb_funcall(trans, consume_bang_id, 1, size);
0
+}
0
+
0
+// Clears size bytes from the transport's string buffer
0
+static bool consume(decode_buffer* buf, int32_t size) {
0
   if (size != 0) {
0
-    rb_funcall(buf->trans, consume_bang_id, 1, INT2FIX(size));
0
+    VALUE ret;
0
+    VALUE args = rb_ary_new3(2, buf->trans, INT2FIX(size));
0
+    int status = 0;
0
+    
0
+    ret = rb_protect(protectable_consume, args, &status);
0
+    
0
+    if (status) {
0
+      return false;
0
+    } else {
0
+      return true;
0
+    }
0
   }
0
+  
0
+  // Nothing to consume, we're all good
0
+  return true;
0
 }
0
 
0
-static VALUE borrow(decode_buffer* buf, int32_t size) {
0
+// This prototype is required to be able to run a call through rb_protect
0
+// which rescues from ruby exceptions
0
+static VALUE protectable_borrow(VALUE args) {
0
+  VALUE trans = rb_ary_entry(args, 0);
0
+  
0
+  switch(RARRAY(args)->len) {
0
+    case 1:
0
+      return rb_funcall(trans, borrow_id, 0);
0
+    
0
+    case 2: {
0
+      VALUE size = rb_ary_entry(args, 1);
0
+      return rb_funcall(trans, borrow_id, 1, size);
0
+    }
0
+  }
0
+  
0
+  return Qnil;
0
+}
0
+
0
+// Calls into the transport to get the available string buffer
0
+static bool borrow(decode_buffer* buf, int32_t size, VALUE* dst) {
0
+  int status = 0;
0
+  VALUE args;
0
+  
0
   if (size == 0) {
0
-    return rb_funcall(buf->trans, borrow_id, 0);
0
+    args = rb_ary_new3(1, buf->trans);
0
   } else {
0
-    return rb_funcall(buf->trans, borrow_id, 1, INT2FIX(size));
0
+    args = rb_ary_new3(2, buf->trans, INT2FIX(size));
0
   }
0
+
0
+  *dst = rb_protect(protectable_borrow, args, &status);
0
+  
0
+  return (status == 0);
0
 }
0
 
0
 // Refills the buffer by calling borrow. If buf->pos is nonzero that number of bytes
0
-// is cleared through consume
0
-static void fill_buffer(decode_buffer* buf, int32_t req_len) {
0
-  consume(buf, buf->pos);
0
-  VALUE refill = borrow(buf, req_len);
0
+// is cleared through consume.
0
+//
0
+// returns: 0 on success, non-zero on failure. On error buf is unchanged.
0
+static int fill_buffer(decode_buffer* buf, int32_t req_len) {
0
+  VALUE refill;
0
+  
0
+  if (!consume(buf, buf->pos)) {
0
+    return -1;
0
+  }
0
+
0
+  if (!borrow(buf, req_len, &refill)) {
0
+    return -2;
0
+  }
0
+  
0
   buf->data = StringValuePtr(refill);
0
   buf->len = RSTRING(refill)->len;
0
   buf->pos = 0;
0
+  
0
+  return 0;
0
 }
0
 
0
 
0
@@ -569,7 +630,10 @@
0
       buf->pos += avail;
0
     }
0
 
0
-    fill_buffer(buf, size);
0
+    if (fill_buffer(buf, size - avail) < 0) {
0
+      return false;
0
+    }
0
+    
0
     memcpy(dst + avail, buf->data, size - avail);
0
     buf->pos += size - avail;
0
   }
0
0
0
0
0
0
0
0
0
0
0
0
0
0
0
0
0
0
0
0
@@ -581,97 +645,112 @@
0
 // Helpers for grabbing specific types from the buffer
0
 // -----------------------------------------------------------------------------
0
 
0
-static int8_t read_byte(decode_buffer* buf) {
0
-  int8_t data;
0
-  read_bytes(buf, &data, sizeof(int8_t));
0
-  return data;
0
+static bool read_byte(decode_buffer* buf, int8_t* data) {
0
+  return read_bytes(buf, data, sizeof(int8_t));
0
 }
0
 
0
-static int16_t read_int16(decode_buffer* buf) {
0
-  int16_t data;
0
-  read_bytes(buf, &data, sizeof(int16_t));
0
-  return ntohs(data);
0
+static bool read_int16(decode_buffer* buf, int16_t* data) {
0
+  bool success = read_bytes(buf, data, sizeof(int16_t));
0
+  *data = ntohs(*data);
0
+  
0
+  return success;
0
 }
0
 
0
-static int32_t read_int32(decode_buffer* buf) {
0
-  int32_t data;
0
-  read_bytes(buf, &data, sizeof(int32_t));
0
-  return ntohl(data);
0
+static bool read_int32(decode_buffer* buf, int32_t* data) {
0
+  bool success = read_bytes(buf, data, sizeof(int32_t));
0
+  *data = ntohl(*data);
0
+  
0
+  return success;
0
 }
0
 
0
-static int64_t read_int64(decode_buffer* buf) {
0
-  int64_t data;
0
-  read_bytes(buf, &data, sizeof(int64_t));
0
-  return ntohll(data);
0
+static bool read_int64(decode_buffer* buf, int64_t* data) {
0
+  bool success = read_bytes(buf, data, sizeof(int64_t));
0
+  *data = ntohll(*data);
0
+  
0
+  return success;
0
 }
0
 
0
-static double read_double(decode_buffer* buf) {
0
-  union {
0
-    double f;
0
-    int64_t t;
0
-  } transfer;
0
-  
0
-  transfer.t = read_int64(buf);
0
-  return transfer.f;
0
+static bool read_double(decode_buffer* buf, double* data) {
0
+  return read_int64(buf, (int64_t*)data);
0
 }
0
 
0
-static VALUE read_string(decode_buffer* buf) {
0
-  VALUE ret_str;
0
-  int len = read_int32(buf);
0
+static bool read_string(decode_buffer* buf, VALUE* data) {
0
+  int len;
0
   
0
+  if (!read_int32(buf, &len)) {
0
+    return false;
0
+  }
0
+  
0
   if (buf->len - buf->pos >= len) {
0
-    ret_str = rb_str_new(buf->data + buf->pos, len);
0
+    *data = rb_str_new(buf->data + buf->pos, len);
0
     buf->pos += len;
0
   } 
0
   else {
0
     char* str;
0
     
0
-    str = (char*) malloc(len);
0
+    if ((str = (char*) malloc(len)) == NULL) {
0
+      return false;
0
+    }
0
     
0
-    read_bytes(buf, str, len);
0
-    ret_str = rb_str_new(str, len);
0
+    if (!read_bytes(buf, str, len)) {
0
+      free(str);
0
+      return false;
0
+    }
0
     
0
+    *data = rb_str_new(str, len);
0
+    
0
     free(str);
0
   }
0
   
0
-  return ret_str;
0
+  return true;
0
 }
0
 
0
-static void read_field_begin(decode_buffer* buf, field_header* header) {
0
+static bool read_field_begin(decode_buffer* buf, field_header* header) {
0
 #ifdef __DEBUG__ // No need for this in prod since I set all the fields
0
   bzero(header, sizeof(field_header));
0
 #endif
0
 
0
   header->name = NULL;
0
-  header->type = read_byte(buf);
0
+  
0
+  if (!read_byte(buf, &header->type)) {
0
+    return false;
0
+  }
0
+  
0
   if (header->type == T_STOP) {
0
     header->id = 0;
0
   } else {
0
-    header->id = read_int16(buf);
0
+    if (!read_int16(buf, &header->id)) {
0
+      return false;
0
+    }
0
   }
0
+  
0
+  return true;
0
 }
0
 
0
 #define read_field_end(buf)
0
 
0
-static void read_map_begin(decode_buffer* buf, map_header* header) {
0
+static bool read_map_begin(decode_buffer* buf, map_header* header) {
0
 #ifdef __DEBUG__ // No need for this in prod since I set all the fields
0
   bzero(header, sizeof(map_header));
0
 #endif
0
   
0
-  header->key_type = read_byte(buf);
0
-  header->val_type = read_byte(buf);
0
-  header->num_entries = read_int32(buf);
0
+  return (read_byte(buf, &header->key_type) && 
0
+          read_byte(buf, &header->val_type) &&
0
+          read_int32(buf, &header->num_entries));
0
 }
0
 
0
 #define read_map_end(buf)
0
 
0
-static void read_list_begin(decode_buffer* buf, list_header* header) {
0
+static bool read_list_begin(decode_buffer* buf, list_header* header) {
0
 #ifdef __DEBUG__ // No need for this in prod since I set all the fields
0
   bzero(header, sizeof(list_header));
0
 #endif
0
   
0
-  header->type = read_byte(buf);
0
-  header->num_elements = read_int32(buf);
0
+  if (!read_byte(buf, &header->type) || !read_int32(buf, &header->num_elements)) {
0
+    return false;
0
+  } else {
0
+    return true;
0
+  }
0
 }
0
 
0
 #define read_list_end(buf)
0
0
0
0
0
0
0
0
0
0
0
0
0
@@ -681,46 +760,101 @@
0
 
0
 
0
 // High level reader function with ruby type coercion
0
-static VALUE read_type(int type, decode_buffer* buf) {
0
+static bool read_type(int type, decode_buffer* buf, VALUE* dst) {
0
   switch(type) {
0
     case T_BOOL: {
0
-      int8_t byte = read_byte(buf);
0
+      int8_t byte;
0
+      
0
+      if (!read_byte(buf, &byte)) {
0
+        return false;
0
+      }
0
+      
0
       if (0 == byte) {
0
-        return Qfalse;
0
+        *dst = Qfalse;
0
       } else {
0
-        return Qtrue;
0
+        *dst = Qtrue;
0
       }
0
+      
0
+      break;
0
     }
0
     
0
-    case T_BYTE:
0
-      return INT2FIX(read_byte(buf));
0
+    case T_BYTE: {
0
+      int8_t byte;
0
+
0
+      if (!read_byte(buf, &byte)) {
0
+        return false;
0
+      }
0
+
0
+      *dst = INT2FIX(byte);
0
+      break;
0
+    }
0
     
0
-    case T_I16:
0
-      return INT2FIX(read_int16(buf));
0
+    case T_I16: {
0
+      int16_t i16;
0
+      
0
+      if (!read_int16(buf, &i16)) {
0
+        return false;
0
+      }
0
+      
0
+      *dst = INT2FIX(i16);
0
+      break;
0
+    }
0
 
0
-    case T_I32:
0
-      return INT2NUM(read_int32(buf));
0
+    case T_I32: {
0
+      int32_t i32;
0
+      
0
+      if (!read_int32(buf, &i32)) {
0
+        return false;
0
+      }
0
+      
0
+      *dst = INT2NUM(i32);
0
+      break;
0
+    }
0
 
0
-    case T_I64:
0
-      return rb_ll2inum(read_int64(buf));
0
+    case T_I64: {
0
+      int64_t i64;
0
+      
0
+      if (!read_int64(buf, &i64)) {
0
+        return false;
0
+      }
0
+      
0
+      *dst = rb_ll2inum(i64);
0
+      break;
0
+    }
0
     
0
-    case T_DBL:
0
-      return rb_float_new(read_double(buf));
0
+    case T_DBL: {
0
+      double dbl;
0
+      
0
+      if (!read_double(buf, &dbl)) {
0
+        return false;
0
+      }
0
+      
0
+      *dst = rb_float_new(dbl);
0
+      break;
0
+    }
0
 
0
     case T_STR: {
0
-      return read_string(buf);
0
+      VALUE str;
0
+      
0
+      if (!read_string(buf, &str)) {
0
+        return false;
0
+      }
0
+      
0
+      *dst = str;
0
+      break;
0
     }
0
   }
0
   
0
-  return Qnil;
0
+  return true;
0
 }
0
 
0
 // TODO(kevinclark): Now that read_string does a malloc,
0
 // This maybe could be modified to avoid that, and the type coercion
0
 
0
 // Read the bytes but don't do anything with the value
0
-static void skip_type(int type, decode_buffer* buf) {
0
-  read_type(type, buf);
0
+static bool skip_type(int type, decode_buffer* buf) {
0
+  VALUE v;
0
+  return read_type(type, buf, &v);
0
 }
0
 
0
 
0
0
@@ -728,11 +862,13 @@
0
 
0
 // Read the right thing from the buffer given the field spec
0
 // and return the ruby object
0
-static VALUE read_field(decode_buffer* buf, field_spec* spec) {
0
+static bool read_field(decode_buffer* buf, field_spec* spec, VALUE* dst) {
0
   switch (spec->type) {
0
     case T_STRCT: {
0
       VALUE obj = rb_class_new_instance(0, NULL, spec->data.class);
0
-      return read_struct(obj, buf);
0
+      
0
+      *dst = read_struct(obj, buf);
0
+      break;
0
     }
0
     
0
     case T_MAP: {
0
0
0
0
0
0
@@ -744,35 +880,49 @@
0
       hsh = rb_hash_new();
0
       
0
       for (i = 0; i < hdr.num_entries; ++i) {
0
-        VALUE key = read_field(buf, spec->data.map->key);
0
-        VALUE val = read_field(buf, spec->data.map->value);
0
+        VALUE key, val;
0
+        
0
+        if (!read_field(buf, spec->data.map->key, &key)) {
0
+          return false;
0
+        }
0
+        
0
+        if (!read_field(buf, spec->data.map->value, &val)) {
0
+          return false;
0
+        }
0
+        
0
         rb_hash_aset(hsh, key, val);
0
       }
0
       
0
       read_map_end(buf);
0
       
0
-      return hsh;
0
+      *dst = hsh;
0
+      break;
0
     }
0
     
0
     case T_LIST: {
0
       list_header hdr;
0
-      VALUE arr;
0
+      VALUE arr, element;
0
       int i;
0
       
0
       read_list_begin(buf, &hdr);
0
       arr = rb_ary_new2(hdr.num_elements);
0
       
0
       for (i = 0; i < hdr.num_elements; ++i) {
0
-        rb_ary_push(arr, read_field(buf, spec->data.element));
0
+        if (!read_field(buf, spec->data.element, &element)) {
0
+          return false;
0
+        }
0
+
0
+        rb_ary_push(arr, element);
0
       }
0
       
0
       read_list_end(buf);
0
       
0
-      return arr;
0
+      *dst = arr;
0
+      break;
0
     }
0
     
0
     case T_SET: {
0
-      VALUE set;
0
+      VALUE set, item;
0
       set_header hdr;
0
       int i;
0
       
0
0
0
0
0
@@ -780,18 +930,36 @@
0
       set = rb_hash_new();
0
       
0
       for (i = 0; i < hdr.num_elements; ++i) {
0
-        rb_hash_aset(set, read_field(buf, spec->data.element), Qtrue);
0
+        if (!read_field(buf, spec->data.element, &item)) {
0
+          return false;
0
+        }
0
+        
0
+        rb_hash_aset(set, item, Qtrue);
0
       }
0
       
0
-      return set;
0
+      *dst = set;
0
+      break;
0
     }
0
     
0
     
0
     default:
0
-      return read_type(spec->type, buf);
0
+      return read_type(spec->type, buf, dst);
0
   }
0
+  
0
+  return true;
0
 }
0
 
0
+static void handle_read_error() {
0
+  // If it was an exception, reraise
0
+  if (!NIL_P(ruby_errinfo)) {
0
+    rb_exc_raise(ruby_errinfo);
0
+  } else {
0
+    // Something else went wrong, no idea what would call this yet
0
+    // So far, the only thing to cause failures underneath is ruby
0
+    // exceptions. Follow up on this regularly -- Kevin Clark (TODO)
0
+    rb_raise(rb_eStandardError, "[BUG] Something went wrong in the field reading, but not a ruby exception");
0
+  }
0
+}
0
 
0
 // Fill in the instance variables in an object (thrift struct)
0
 // from the decode buffer
0
@@ -806,7 +974,10 @@
0
   read_struct_begin(buf);
0
   
0
   while(true) {
0
-    read_field_begin(buf, &f_header);
0
+    if (!read_field_begin(buf, &f_header)) {
0
+      handle_read_error();
0
+    }
0
+    
0
     if (T_STOP == f_header.type) {
0
       break;
0
     }
0
0
0
@@ -814,15 +985,28 @@
0
     field = rb_hash_aref(fields, INT2FIX(f_header.id));
0
     
0
     if (NIL_P(field)) {
0
-      skip_type(f_header.type, buf);
0
+      if (!skip_type(f_header.type, buf)) {
0
+        handle_read_error();
0
+        return Qnil;
0
+      }
0
     } 
0
     else {
0
       spec = parse_field_spec(field);
0
 
0
       if (spec->type != f_header.type) {
0
-        skip_type(spec->type, buf);
0
+        if (!skip_type(spec->type, buf)) {
0
+          free_field_spec(spec);
0
+          handle_read_error();
0
+          return Qnil;
0
+        }
0
       } else {
0
-        value = read_field(buf, spec);
0
+        // Read busted somewhere (probably borrow/consume), bail
0
+        if (!read_field(buf, spec, &value)) {
0
+          free_field_spec(spec);
0
+          handle_read_error();
0
+          return Qnil;
0
+        }
0
+        
0
         name_buf[0] = '@';
0
         strlcpy(&name_buf[1], spec->name, sizeof(name_buf) - 1);
0
         
0
@@ -849,7 +1033,11 @@
0
 
0
   buf.pos = 0;    // This needs to be set so an arbitrary number of bytes isn't consumed
0
   buf.trans = transport;       // We need to hold this so the buffer can be refilled
0
-  fill_buffer(&buf, 0);
0
+  
0
+  if (fill_buffer(&buf, 0) < 0) {
0
+    handle_read_error();
0
+    return Qnil;
0
+  }
0
 
0
 #ifdef __DEBUG__
0
   rb_p(rb_str_new2("Running decode binary with data:"));
0
0
0
0
@@ -880,21 +1068,35 @@
0
   
0
   buf.pos = 0;              // This needs to be set so fill_buffer doesn't consume
0
   buf.trans = trans;        // We need to hold this so the buffer can be refilled
0
-  fill_buffer(&buf, 0);
0
+  
0
 
0
-  version = read_int32(&buf);
0
+  if (fill_buffer(&buf, 0) < 0 || !read_int32(&buf, &version)) {
0
+    // Consume whatever was read
0
+    consume(&buf, buf.pos);
0
+    handle_read_error();
0
+    return Qnil;
0
+  }
0
   
0
   if ((version & VERSION_MASK) != VERSION_1) {
0
-    ID tprotocol_exception = rb_intern("TProtocolException");
0
+    VALUE tprotocol_exception = rb_const_get(rb_cObject, rb_intern("TProtocolException"));
0
     VALUE exception = rb_funcall(tprotocol_exception, rb_intern("new"), 2, rb_const_get(tprotocol_exception, rb_intern("BAD_VERSION")), rb_str_new2("Missing version identifier"));
0
     rb_raise(exception, "");
0
   }
0
   
0
   type = version & 0x000000ff;
0
-  name = read_string(&buf);
0
-  seqid = read_int32(&buf);
0
-
0
-  consume(&buf, buf.pos);
0
+  
0
+  if (!read_string(&buf, &name) || !read_int32(&buf, &seqid)) {
0
+    // Consume whatever was read
0
+    consume(&buf, buf.pos);
0
+    handle_read_error();
0
+    return Qnil;
0
+  }
0
+  
0
+  // Consume whatever was read
0
+  if (consume(&buf, buf.pos) < 0) {
0
+    handle_read_error();
0
+    return Qnil;
0
+  }
0
   
0
   return rb_ary_new3(3, name, INT2FIX(type), INT2FIX(seqid));
0
 }

0
@@ -2,7 +2,7 @@
0
 require 'tbinaryprotocolaccelerated'
0
 
0
 =begin
0
-The only change required for a transport to support TFastBinaryProtocol is to implement 2 methods:
0
+The only change required for a transport to support TBinaryProtocolAccelerated is to implement 2 methods:
0
   * borrow(size), which takes an optional argument and returns atleast _size_ bytes from the transport, 
0
                   or the default buffer size if no argument is given
0
   * consume!(size), which removes size bytes from the front of the buffer