A few minor changes to the streaming protocol.

1. the start and end callbacks can now return
   a upb_flow_t and set a status message.
2. clarified some semantics around passing an
   error status back from the callbacks.

5 files changed, 76 insertions, 59 deletions

diff --git a/core/upb.c b/core/upb.c
index 2f715d0..05e9b7d 100644
--- a/core/upb.c
+++ b/core/upb.c
@@ -73,3 +73,7 @@ void upb_printerr(upb_status *status) {
     fprintf(stderr, "code: %d, no msg\n", status->code);
   }
 }
+
+void upb_status_uninit(upb_status *status) {
+  upb_string_unref(status->str);
+}
diff --git a/core/upb.h b/core/upb.h
index 764e9ba..fb6d9ea 100644
--- a/core/upb.h
+++ b/core/upb.h
@@ -290,30 +290,27 @@ INLINE void upb_value_write(upb_valueptr ptr, upb_value val,
 // Status codes used as a return value.  Codes >0 are not fatal and can be
 // resumed.
 enum upb_status_code {
+  // The operation completed successfully.
   UPB_STATUS_OK = 0,
 
-  // A read or write from a streaming src/sink could not be completed right now.
-  UPB_STATUS_TRYAGAIN = 1,
+  // The bytesrc is at EOF and all data was read successfully.
+  UPB_STATUS_EOF = 1,
 
-  // A value had an incorrect wire type and will be skipped.
-  UPB_STATUS_BADWIRETYPE = 2,
+  // A read or write from a streaming src/sink could not be completed right now.
+  UPB_STATUS_TRYAGAIN = 2,
 
   // An unrecoverable error occurred.
   UPB_STATUS_ERROR = -1,
 
-  // A varint went for 10 bytes without terminating.
-  UPB_ERROR_UNTERMINATED_VARINT = -2,
-
-  // The max nesting level (UPB_MAX_NESTING) was exceeded.
-  UPB_ERROR_MAX_NESTING_EXCEEDED = -3
+  // A recoverable error occurred (for example, data of the wrong type was
+  // encountered which we can skip over).
+  // UPB_STATUS_RECOVERABLE_ERROR = -2
 };
 
-// TODO: consider making this a single word: a upb_string* where we use the low
-// bits as flags indicating whether there is an error and whether it is
-// resumable.  This would improve efficiency, because the code would not need
-// to be loaded after a call to a function returning a status.
+// TODO: consider adding error space and code, to let ie. errno be stored
+// as a proper code.
 typedef struct {
-  enum upb_status_code code;
+  char code;
   upb_string *str;
 } upb_status;
 
@@ -329,6 +326,8 @@ INLINE void upb_status_init(upb_status *status) {
   status->str = NULL;
 }
 
+void upb_status_uninit(upb_status *status);
+
 void upb_printerr(upb_status *status);
 void upb_clearerr(upb_status *status);
 void upb_seterr(upb_status *status, enum upb_status_code code, const char *msg,
diff --git a/core/upb_def.c b/core/upb_def.c
index 4f12dbe..0176dc9 100644
--- a/core/upb_def.c
+++ b/core/upb_def.c
@@ -257,6 +257,7 @@ struct _upb_defbuilder {
   upb_deflist defs;
   upb_defbuilder_frame stack[UPB_MAX_TYPE_DEPTH];
   int stack_len;
+  upb_status status;
 
   uint32_t number;
   upb_string *name;
@@ -275,12 +276,14 @@ static void upb_enumdef_register_EnumDescriptorProto(upb_defbuilder *b,
 
 static void upb_defbuilder_init(upb_defbuilder *b) {
   upb_deflist_init(&b->defs);
+  upb_status_init(&b->status);
   b->stack_len = 0;
   b->name = NULL;
 }
 
 static void upb_defbuilder_uninit(upb_defbuilder *b) {
   upb_string_unref(b->name);
+  upb_status_uninit(&b->status);
   upb_deflist_uninit(&b->defs);
 }
 
@@ -356,7 +359,7 @@ static void upb_defbuilder_register_FileDescriptorProto(upb_defbuilder *b,
     &upb_defbuilder_FileDescriptorProto_startsubmsg,
   };
   upb_register_handlerset(h, &handlers);
-  upb_set_handler_closure(h, b);
+  upb_set_handler_closure(h, b, &b->status);
 }
 
 // Handlers for google.protobuf.FileDescriptorSet.
@@ -392,7 +395,7 @@ static void upb_defbuilder_register_FileDescriptorSet(
     &upb_defbuilder_FileDescriptorSet_startsubmsg,
   };
   upb_register_handlerset(h, &handlers);
-  upb_set_handler_closure(h, b);
+  upb_set_handler_closure(h, b, &b->status);
 }
 
 
@@ -439,10 +442,11 @@ static void upb_enumdef_free(upb_enumdef *e) {
 }
 
 // google.protobuf.EnumValueDescriptorProto.
-static void upb_enumdef_EnumValueDescriptorProto_startmsg(void *_b) {
+static upb_flow_t upb_enumdef_EnumValueDescriptorProto_startmsg(void *_b) {
   upb_defbuilder *b = _b;
   b->saw_number = false;
   b->saw_name = false;
+  return UPB_CONTINUE;
 }
 
 static upb_flow_t upb_enumdef_EnumValueDescriptorProto_value(void *_b,
@@ -463,12 +467,12 @@ static upb_flow_t upb_enumdef_EnumValueDescriptorProto_value(void *_b,
   return UPB_CONTINUE;
 }
 
-static void upb_enumdef_EnumValueDescriptorProto_endmsg(void *_b) {
+static upb_flow_t upb_enumdef_EnumValueDescriptorProto_endmsg(void *_b) {
   upb_defbuilder *b = _b;
   if(!b->saw_number || !b->saw_name) {
-    //upb_seterr(status, UPB_STATUS_ERROR, "Enum value missing name or number.");
-    //goto err;
-    return;
+    upb_seterr(&b->status, UPB_STATUS_ERROR,
+               "Enum value missing name or number.");
+    return UPB_STOP;
   }
   upb_ntoi_ent ntoi_ent = {{b->name, 0}, b->number};
   upb_iton_ent iton_ent = {{b->number, 0}, b->name};
@@ -478,6 +482,7 @@ static void upb_enumdef_EnumValueDescriptorProto_endmsg(void *_b) {
   // We don't unref "name" because we pass our ref to the iton entry of the
   // table.  strtables can ref their keys, but the inttable doesn't know that
   // the value is a string.
+  return UPB_CONTINUE;
 }
 
 static void upb_enumdef_register_EnumValueDescriptorProto(upb_defbuilder *b,
@@ -488,22 +493,24 @@ static void upb_enumdef_register_EnumValueDescriptorProto(upb_defbuilder *b,
     &upb_enumdef_EnumValueDescriptorProto_value,
   };
   upb_register_handlerset(h, &handlers);
-  upb_set_handler_closure(h, b);
+  upb_set_handler_closure(h, b, &b->status);
 }
 
 // google.protobuf.EnumDescriptorProto.
-void upb_enumdef_EnumDescriptorProto_startmsg(void *_b) {
+static upb_flow_t upb_enumdef_EnumDescriptorProto_startmsg(void *_b) {
   upb_defbuilder *b = _b;
   upb_enumdef *e = malloc(sizeof(*e));
   upb_def_init(&e->base, UPB_DEF_ENUM);
   upb_strtable_init(&e->ntoi, 0, sizeof(upb_ntoi_ent));
   upb_inttable_init(&e->iton, 0, sizeof(upb_iton_ent));
   upb_deflist_push(&b->defs, UPB_UPCAST(e));
+  return UPB_CONTINUE;
 }
 
-void upb_enumdef_EnumDescriptorProto_endmsg(void *_b) {
+static upb_flow_t upb_enumdef_EnumDescriptorProto_endmsg(void *_b) {
   upb_defbuilder *b = _b;
   assert(upb_defbuilder_last(b)->fqname != NULL);
+  return UPB_CONTINUE;
 }
 
 static upb_flow_t upb_enumdef_EnumDescriptorProto_value(void *_b,
@@ -546,7 +553,7 @@ static void upb_enumdef_register_EnumDescriptorProto(upb_defbuilder *b,
     &upb_enumdef_EnumDescriptorProto_startsubmsg,
   };
   upb_register_handlerset(h, &handlers);
-  upb_set_handler_closure(h, b);
+  upb_set_handler_closure(h, b, &b->status);
 }
 
 upb_enum_iter upb_enum_begin(upb_enumdef *e) {
@@ -576,7 +583,7 @@ static void upb_fielddef_free(upb_fielddef *f) {
   free(f);
 }
 
-static void upb_fielddef_startmsg(void *_b) {
+static upb_flow_t upb_fielddef_startmsg(void *_b) {
   upb_defbuilder *b = _b;
   upb_fielddef *f = malloc(sizeof(*f));
   f->number = -1;
@@ -585,9 +592,10 @@ static void upb_fielddef_startmsg(void *_b) {
   f->owned = false;
   f->msgdef = upb_defbuilder_top(b);
   b->f = f;
+  return UPB_CONTINUE;
 }
 
-static void upb_fielddef_endmsg(void *_b) {
+static upb_flow_t upb_fielddef_endmsg(void *_b) {
   upb_defbuilder *b = _b;
   upb_fielddef *f = b->f;
   // TODO: verify that all required fields were present.
@@ -600,6 +608,7 @@ static void upb_fielddef_endmsg(void *_b) {
   upb_ntof_ent ntof_ent = {{f->name, 0}, f};
   upb_inttable_insert(&m->itof, &itof_ent.e);
   upb_strtable_insert(&m->ntof, &ntof_ent.e);
+  return UPB_CONTINUE;
 }
 
 static upb_flow_t upb_fielddef_value(void *_b, upb_fielddef *f, upb_value val) {
@@ -620,7 +629,7 @@ static upb_flow_t upb_fielddef_value(void *_b, upb_fielddef *f, upb_value val) {
       break;
     case GOOGLE_PROTOBUF_FIELDDESCRIPTORPROTO_TYPE_NAME_FIELDNUM: {
       upb_string *str = upb_string_new();
-      if (!upb_value_getfullstr(val, str, NULL)) return UPB_ERROR;
+      if (!upb_value_getfullstr(val, str, NULL)) return UPB_STOP;
       if(b->f->def) upb_def_unref(b->f->def);
       b->f->def = UPB_UPCAST(upb_unresolveddef_new(str));
       b->f->owned = true;
@@ -638,7 +647,7 @@ static void upb_fielddef_register_FieldDescriptorProto(upb_defbuilder *b,
     &upb_fielddef_value,
   };
   upb_register_handlerset(h, &handlers);
-  upb_set_handler_closure(h, b);
+  upb_set_handler_closure(h, b, &b->status);
 }
 
 
@@ -658,7 +667,7 @@ static int upb_compare_fields(const void *f1, const void *f2) {
 }
 
 // google.protobuf.DescriptorProto.
-static void upb_msgdef_startmsg(void *_b) {
+static upb_flow_t upb_msgdef_startmsg(void *_b) {
   upb_defbuilder *b = _b;
   upb_msgdef *m = malloc(sizeof(*m));
   upb_def_init(&m->base, UPB_DEF_MSG);
@@ -667,15 +676,16 @@ static void upb_msgdef_startmsg(void *_b) {
   upb_strtable_init(&m->ntof, 4, sizeof(upb_ntof_ent));
   upb_deflist_push(&b->defs, UPB_UPCAST(m));
   upb_defbuilder_startcontainer(b);
+  return UPB_CONTINUE;
 }
 
-static void upb_msgdef_endmsg(void *_b) {
+static upb_flow_t upb_msgdef_endmsg(void *_b) {
   upb_defbuilder *b = _b;
   upb_msgdef *m = upb_defbuilder_top(b);
   if(!m->base.fqname) {
-    //upb_seterr(status, UPB_STATUS_ERROR, "Encountered message with no name.");
-    //return UPB_ERROR;
-    return;
+    upb_seterr(&b->status, UPB_STATUS_ERROR,
+               "Encountered message with no name.");
+    return UPB_STOP;
   }
 
   // Create an ordering over the fields.
@@ -716,7 +726,7 @@ static void upb_msgdef_endmsg(void *_b) {
   if (max_align > 0) m->size = upb_align_up(m->size, max_align);
 
   upb_defbuilder_endcontainer(b);
-  //return UPB_CONTINUE;
+  return UPB_CONTINUE;
 }
 
 static upb_flow_t upb_msgdef_value(void *_b, upb_fielddef *f, upb_value val) {
@@ -767,7 +777,7 @@ static void upb_msgdef_register_DescriptorProto(upb_defbuilder *b,
     &upb_msgdef_startsubmsg,
   };
   upb_register_handlerset(h, &handlers);
-  upb_set_handler_closure(h, b);
+  upb_set_handler_closure(h, b, &b->status);
 }
 
 static void upb_msgdef_free(upb_msgdef *m)
@@ -1100,16 +1110,14 @@ void upb_symtab_addfds(upb_symtab *s, upb_src *src, upb_status *status)
 {
   upb_defbuilder b;
   upb_defbuilder_init(&b);
-  //upb_defbuilder_register_FileDescriptorSet(&b, upb_src_gethandlers(src));
-  upb_defbuilder_register_FileDescriptorSet(&b, NULL);
-  if(!upb_src_run(src)) {
-    upb_copyerr(status, upb_src_status(src));
-    upb_defbuilder_uninit(&b);
-    return;
-  }
-  upb_symtab_add_defs(s, b.defs.defs, b.defs.len, false, status);
+  upb_handlers handlers;
+  upb_handlers_init(&handlers);
+  upb_defbuilder_register_FileDescriptorSet(&b, &handlers);
+  upb_src_sethandlers(src, &handlers);
+  upb_src_run(src, status);
+  if (upb_ok(status))
+    upb_symtab_add_defs(s, b.defs.defs, b.defs.len, false, status);
   upb_defbuilder_uninit(&b);
-  return;
 }
 
 
diff --git a/core/upb_stream.h b/core/upb_stream.h
index 9ae69de..40836e9 100644
--- a/core/upb_stream.h
+++ b/core/upb_stream.h
@@ -39,12 +39,8 @@ typedef enum {
   // Caller should continue sending values to the sink.
   UPB_CONTINUE,
 
-  // An error occurred; check status for details.
-  UPB_ERROR,
-
-  // Processing should stop for now, but could be resumed later.
-  // If processing resumes later, it should resume with the next value.
-  UPB_SUSPEND,
+  // Stop processing for now; check status for details.
+  UPB_STOP,
 
   // Skips to the end of the current submessage (or if we are at the top
   // level, skips to the end of the entire message).
@@ -61,8 +57,8 @@ typedef enum {
 struct _upb_handlers;
 typedef struct _upb_handlers upb_handlers;
 
-typedef void (*upb_startmsg_handler_t)(void *closure);
-typedef void (*upb_endmsg_handler_t)(void *closure);
+typedef upb_flow_t (*upb_startmsg_handler_t)(void *closure);
+typedef upb_flow_t (*upb_endmsg_handler_t)(void *closure);
 typedef upb_flow_t (*upb_value_handler_t)(void *closure,
                                           struct _upb_fielddef *f,
                                           upb_value val);
@@ -76,12 +72,14 @@ typedef upb_flow_t (*upb_unknownval_handler_t)(void *closure,
 
 // An empty set of handlers, for convenient copy/paste:
 //
-// static void startmsg(void *closure) {
+// static upb_flow_t startmsg(void *closure) {
 //   // Called when the top-level message begins.
+//   return UPB_CONTINUE;
 // }
 //
-// static void endmsg(void *closure) {
+// static upb_flow_t endmsg(void *closure) {
 //   // Called when the top-level message ends.
+//   return UPB_CONTINUE;
 // }
 //
 // static upb_flow_t value(void *closure, upb_fielddef *f, upb_value val) {
@@ -120,10 +118,15 @@ INLINE void upb_handlers_uninit(upb_handlers *h);
 INLINE void upb_handlers_reset(upb_handlers *h);
 INLINE bool upb_handlers_isempty(upb_handlers *h);
 INLINE void upb_register_handlerset(upb_handlers *h, upb_handlerset *set);
+
 // TODO: for clients that want to increase efficiency by preventing bytesrcs
 // from automatically being converted to strings in the value callback.
 // INLINE void upb_handlers_use_bytesrcs(bool use_bytesrcs);
-INLINE void upb_set_handler_closure(upb_handlers *h, void *closure);
+
+// The closure will be passed to every handler.  The status will be used
+// only immediately after a handler has returned UPB_STOP.
+INLINE void upb_set_handler_closure(upb_handlers *h, void *closure,
+                                    upb_status *status);
 
 // An object that transparently handles delegation so that the caller needs
 // only follow the protocol as if delegation did not exist.
@@ -146,8 +149,8 @@ INLINE upb_flow_t upb_dispatch_unknownval(upb_dispatcher *d,
 struct _upb_src;
 typedef struct _upb_src upb_src;
 
-bool upb_src_run(upb_src *src);
-upb_status *upb_src_status(upb_src *src);
+void upb_src_sethandlers(upb_src *src, upb_handlers *handlers);
+void upb_src_run(upb_src *src, upb_status *status);
 
 
 /* upb_bytesrc ****************************************************************/
diff --git a/core/upb_stream_vtbl.h b/core/upb_stream_vtbl.h
index c0cf04f..d017177 100644
--- a/core/upb_stream_vtbl.h
+++ b/core/upb_stream_vtbl.h
@@ -133,6 +133,7 @@ INLINE upb_status *upb_bytesink_status(upb_bytesink *sink) {
 struct _upb_handlers {
   upb_handlerset *set;
   void *closure;
+  upb_status *status;  // We don't own this.
 };
 
 INLINE void upb_handlers_init(upb_handlers *h) {
@@ -155,8 +156,10 @@ INLINE void upb_register_handlerset(upb_handlers *h, upb_handlerset *set) {
   h->set = set;
 }
 
-INLINE void upb_set_handler_closure(upb_handlers *h, void *closure) {
+INLINE void upb_set_handler_closure(upb_handlers *h, void *closure,
+                                    upb_status *status) {
   h->closure = closure;
+  h->status = status;
 }
 
 // upb_dispatcher