Implemented source for upb_msg.

3 files changed, 39 insertions, 17 deletions

diff --git a/src/upb_data.c b/src/upb_data.c
index 4fc1844..da72250 100644
--- a/src/upb_data.c
+++ b/src/upb_data.c
@@ -300,6 +300,39 @@ void upb_msg_parsestr(upb_msg *msg, struct upb_msgdef *md, upb_strptr str,
 
 /* upb_msgsrc  ****************************************************************/
 
+static void _upb_msgsrc_produceval(union upb_value v, struct upb_fielddef *f,
+                                   upb_sink *sink)
+{
+  if(upb_issubmsg(f)) {
+    upb_sink_onstart(sink, f);
+    upb_msgsrc_produce(v.msg, upb_downcast_msgdef(f->def), sink);
+    upb_sink_onend(sink);
+  } else if(upb_isstring(f)) {
+    upb_sink_onstr(sink, f, v.str, 0, upb_strlen(v.str));
+  } else {
+    upb_sink_onvalue(sink, f, v);
+  }
+}
+
+void upb_msgsrc_produce(upb_msg *msg, struct upb_msgdef *md, upb_sink *sink)
+{
+  for(int i = 0; i < md->num_fields; i++) {
+    struct upb_fielddef *f = &md->fields[i];
+    if(!upb_msg_has(msg, f)) continue;
+    union upb_value v = upb_msg_get(msg, f);
+    if(upb_isarray(f)) {
+      upb_arrayptr arr = v.arr;
+      for(upb_arraylen_t j = 0; j < upb_array_len(arr); j++) {
+        union upb_value elem = upb_array_get(arr, f, j);
+        _upb_msgsrc_produceval(elem, f, sink);
+      }
+    } else {
+      _upb_msgsrc_produceval(v, f, sink);
+    }
+  }
+}
+
+
 /* upb_msgsink  ***************************************************************/
 
 struct upb_msgsink_frame {
diff --git a/src/upb_data.h b/src/upb_data.h
index 4fc9e8d..b0c9b6d 100644
--- a/src/upb_data.h
+++ b/src/upb_data.h
@@ -522,22 +522,9 @@ void upb_msg_parsestr(upb_msg *msg, struct upb_msgdef *md, upb_strptr str,
 
 /* upb_msgsrc *****************************************************************/
 
-// A upb_msgsrc can push the data of a upb_msg to a upb_sink.
-struct upb_msgsrc;
-typedef struct upb_msgsrc upb_msgsrc;
-
-// Allocate and free a msgsrc, respectively.
-upb_msgsrc *upb_msgsrc_new();
-void upb_msgsrc_free(upb_msgsrc *src);
-
-// Resets the msgsrc for the given msg, msgdef, and sink.  This must be
-// called before upb_msgsrc_produce().
-void upb_msgsrc_reset(upb_msgsrc *src, upb_msg *msg, struct upb_msgdef *md,
-                      upb_sink *sink);
-
-// Pushes data from the upb_msgsrc to the sink that was provided at the last
-// reset.  Returns true if the sink is finished, or false if it is suspended.
-bool upb_msgsrc_produce(upb_msgsrc *src);
+// A nonresumable, non-interruptable (but simple and fast) source for pushing
+// the data of a upb_msg to a upb_sink.
+void upb_msgsrc_produce(upb_msg *msg, struct upb_msgdef *md, upb_sink *sink);
 
 
 /* upb_msgsink ****************************************************************/
diff --git a/src/upb_sink.h b/src/upb_sink.h
index 17e1e1d..3400092 100644
--- a/src/upb_sink.h
+++ b/src/upb_sink.h
@@ -44,7 +44,9 @@ typedef enum {
   UPB_SINK_CONTINUE,
 
   // The consumer has consumed the current value, but it wants to stop for now.
-  // When the producer is next invoked, it should resume at the next value.
+  // When the source is next invoked, it should resume at the next value.  Note
+  // that sources are not necessarily resumable; if a source is not resumable,
+  // returning UPB_SINK_SUSPEND will simply halt it.
   UPB_SINK_SUSPEND,
 
   // The consumer wants to skip to the end of the current submessage and