add streming example to docs

Author: svacas

native-lib/README.md

@@ -220,3 +220,34 @@ finally:
     # Optional: if you used the global API and want to force cleanup
     dataweave.cleanup()
 ```
+
+### 6) I/O streaming with callbacks
+
+Use `run_input_output_callback` to feed input and consume output through callbacks, enabling constant-memory processing of arbitrarily large data.
+
+```python
+json_input = b'[1,2,3,4,5]'
+pos = 0
+
+def read_cb(buf_size):
+    nonlocal pos
+    chunk = json_input[pos:pos + buf_size]
+    pos += len(chunk)
+    return chunk  # return b"" when done
+
+chunks = []
+def write_cb(data):
+    chunks.append(data)
+    return 0  # 0 = success
+
+result = dataweave.run_input_output_callback(
+    "output application/json deferred=true --- payload map ($ * $)",
+    input_name="payload",
+    input_mime_type="application/json",
+    read_callback=read_cb,
+    write_callback=write_cb,
+)
+
+print(result)            # {"success": True}
+print(b"".join(chunks))  # [1,4,9,16,25]
+```

native-lib/example_streaming.py

@@ -25,23 +25,30 @@ def example_streaming_input_output_callback():
         input_iter = iter(range(num_elements))
         input_started = False
         input_done = False
+        pending_token = None
 
         def read_callback(buf_size):
-            nonlocal input_started, input_done
+            nonlocal input_started, input_done, pending_token
             if input_done:
                 return b""
             parts = []
             if not input_started:
                 parts.append(b"[")
                 input_started = True
             remaining = buf_size - sum(len(p) for p in parts)
+            if pending_token is not None:
+                if len(pending_token) <= remaining:
+                    parts.append(pending_token)
+                    remaining -= len(pending_token)
+                    pending_token = None
+                else:
+                    return b"".join(parts)
             try:
                 while remaining > 0:
                     i = next(input_iter)
                     token = (b"," if i > 0 else b"") + str(i).encode("utf-8")
                     if len(token) > remaining:
-                        # put it back via a wrapper – simpler: just include it and break
-                        parts.append(token)
+                        pending_token = token
                         break
                     parts.append(token)
                     remaining -= len(token)
@@ -93,8 +100,36 @@ def write_callback(data):
         return False
 
 
+def doc_example():
+    json_input = b'[1,2,3,4,5]'
+    pos = 0
+
+    def read_cb(buf_size):
+        nonlocal pos
+        chunk = json_input[pos:pos + buf_size]
+        pos += len(chunk)
+        return chunk  # return b"" when done
+
+    chunks = []
+    def write_cb(data):
+        chunks.append(data)
+        return 0  # 0 = success
+
+    result = dataweave.run_input_output_callback(
+        "output application/json deferred=true --- payload map ($ * $)",
+        input_name="payload",
+        input_mime_type="application/json",
+        read_callback=read_cb,
+        write_callback=write_cb,
+    )
+
+    print(result)            # {"success": True}
+    print(b"".join(chunks))  # [1,4,9,16,25]
+
+
 def main():
     example_streaming_input_output_callback()
+    doc_example()
 
 
 if __name__ == "__main__":

native-lib/python/src/dataweave/__init__.py

@@ -467,7 +467,7 @@ def _read_cb(_ctx, buf, buf_size):
                 data = read_callback(buf_size)
                 if not data:
                     return 0  # EOF
-                n = len(data)
+                n = min(len(data), buf_size)
                 ctypes.memmove(buf, data, n)
                 return n
             except Exception: