#!/usr/bin/env escript %%% @doc PyBuffer example - demonstrates zero-copy WSGI input buffer. %%% %%% This example shows how to use py_buffer for streaming HTTP body %%% data from Erlang to Python, suitable for WSGI/ASGI input. %%% %%% Prerequisites: rebar3 compile %%% Run from project root: escript examples/py_buffer_example.erl -mode(compile). main(_) -> %% Add the compiled beam files to the code path ScriptDir = filename:dirname(escript:script_name()), ProjectRoot = filename:dirname(ScriptDir), EbinDir = filename:join([ProjectRoot, "_build", "default", "lib", "erlang_python", "ebin"]), true = code:add_pathz(EbinDir), {ok, _} = application:ensure_all_started(erlang_python), io:format("~n=== PyBuffer Zero-Copy WSGI Input Demo ===~n~n"), %% Demo 1: Basic buffer usage basic_buffer_demo(), %% Demo 2: Simulated HTTP body streaming http_body_demo(), %% Demo 3: File-like interface file_like_demo(), %% Demo 4: Zero-copy memoryview access memoryview_demo(), %% Demo 5: Line iteration line_iteration_demo(), %% Demo 6: Async I/O asyncio_demo(), io:format("=== Done ===~n~n"), ok. basic_buffer_demo() -> io:format("--- Basic Buffer Demo ---~n~n"), %% Create a buffer io:format("Creating buffer...~n"), {ok, Buf} = py_buffer:new(), %% Write some data io:format("Writing data chunks...~n"), ok = py_buffer:write(Buf, <<"Hello, ">>), ok = py_buffer:write(Buf, <<"World!">>), %% Close to signal EOF ok = py_buffer:close(Buf), io:format("Buffer closed (EOF signaled)~n"), %% Pass to Python and read io:format("Reading from Python...~n"), Ctx = py:context(1), ok = py:exec(Ctx, <<" def read_all(buf): data = buf.read() print(f' Read {len(data)} bytes: {data}') return data ">>), {ok, Data} = py:eval(Ctx, <<"read_all(buf)">>, #{<<"buf">> => Buf}), io:format("Erlang received: ~p~n~n", [Data]), ok. http_body_demo() -> io:format("--- HTTP Body Streaming Demo ---~n~n"), %% Simulate receiving a JSON POST body Body = <<"{\"user\": \"alice\", \"action\": \"login\", \"timestamp\": 1234567890}">>, ContentLength = byte_size(Body), io:format("Simulating HTTP POST with ~p byte body~n", [ContentLength]), %% Create buffer with known content length (pre-allocates) {ok, Buf} = py_buffer:new(ContentLength), %% Write the body (could be in chunks) ok = py_buffer:write(Buf, Body), ok = py_buffer:close(Buf), %% Build WSGI-like environ Ctx = py:context(1), ok = py:exec(Ctx, <<" import json def handle_request(environ): '''Simulate WSGI request handler.''' method = environ.get('REQUEST_METHOD', 'GET') path = environ.get('PATH_INFO', '/') content_type = environ.get('CONTENT_TYPE', '') print(f' {method} {path}') print(f' Content-Type: {content_type}') # Read body from wsgi.input (PyBuffer) wsgi_input = environ.get('wsgi.input') if wsgi_input: body = wsgi_input.read() print(f' Body ({len(body)} bytes): {body[:50]}...' if len(body) > 50 else f' Body: {body}') if content_type == 'application/json': data = json.loads(body) return {'status': 'ok', 'user': data.get('user')} return {'status': 'ok'} ">>), Environ = #{ <<"REQUEST_METHOD">> => <<"POST">>, <<"PATH_INFO">> => <<"/api/login">>, <<"CONTENT_TYPE">> => <<"application/json">>, <<"CONTENT_LENGTH">> => integer_to_binary(ContentLength), <<"wsgi.input">> => Buf }, {ok, Result} = py:eval(Ctx, <<"handle_request(environ)">>, #{<<"environ">> => Environ}), io:format("Response: ~p~n~n", [Result]), ok. file_like_demo() -> io:format("--- File-Like Interface Demo ---~n~n"), Ctx = py:context(1), %% Create buffer with multiple lines {ok, Buf} = py_buffer:new(), ok = py_buffer:write(Buf, <<"Name: Alice\n">>), ok = py_buffer:write(Buf, <<"Email: alice@example.com\n">>), ok = py_buffer:write(Buf, <<"Role: Admin\n">>), ok = py_buffer:close(Buf), ok = py:exec(Ctx, <<" def demonstrate_file_methods(buf): '''Show file-like methods.''' print(' File-like properties:') print(f' readable(): {buf.readable()}') print(f' writable(): {buf.writable()}') print(f' seekable(): {buf.seekable()}') print(f' len(buf): {len(buf)}') print() # Read first line line1 = buf.readline() print(f' readline(): {line1}') # Current position pos = buf.tell() print(f' tell(): {pos}') # Seek back to start buf.seek(0) print(f' seek(0), tell(): {buf.tell()}') # Read all remaining rest = buf.read() print(f' read(): {rest[:30]}...') return 'done' ">>), {ok, _} = py:eval(Ctx, <<"demonstrate_file_methods(buf)">>, #{<<"buf">> => Buf}), io:format("~n"), ok. memoryview_demo() -> io:format("--- Zero-Copy Memoryview Demo ---~n~n"), Ctx = py:context(1), %% Create buffer with binary data Data = <<"HEADER:12345:PAYLOAD:abcdefghijklmnopqrstuvwxyz:END">>, {ok, Buf} = py_buffer:new(byte_size(Data)), ok = py_buffer:write(Buf, Data), ok = py_buffer:close(Buf), ok = py:exec(Ctx, <<" def zero_copy_parse(buf): '''Demonstrate zero-copy access via memoryview.''' # Get memoryview - no data copying! mv = memoryview(buf) print(f' memoryview created, {len(mv)} bytes') print(f' readonly: {mv.readonly}') print(f' ndim: {mv.ndim}') # Find colon positions using find (uses memchr internally) data_bytes = bytes(mv) # Only for find, still efficient # Parse header first_colon = buf.find(b':') header = bytes(mv[:first_colon]) print(f' Header: {header}') # Find PAYLOAD section payload_start = buf.find(b'PAYLOAD:') + 8 payload_end = buf.find(b':END') payload = bytes(mv[payload_start:payload_end]) print(f' Payload: {payload}') # Release memoryview mv.release() print(' memoryview released') return payload ">>), {ok, Payload} = py:eval(Ctx, <<"zero_copy_parse(buf)">>, #{<<"buf">> => Buf}), io:format("Extracted payload: ~p~n~n", [Payload]), ok. line_iteration_demo() -> io:format("--- Line Iteration Demo ---~n~n"), Ctx = py:context(1), %% Create buffer with CSV-like data {ok, Buf} = py_buffer:new(), ok = py_buffer:write(Buf, <<"id,name,score\n">>), ok = py_buffer:write(Buf, <<"1,Alice,95\n">>), ok = py_buffer:write(Buf, <<"2,Bob,87\n">>), ok = py_buffer:write(Buf, <<"3,Charlie,92\n">>), ok = py_buffer:close(Buf), ok = py:exec(Ctx, <<" def process_csv(buf): '''Iterate over lines like a file.''' records = [] header = None for line in buf: line = line.strip() if not line: continue parts = line.decode().split(',') if header is None: header = parts print(f' Header: {header}') else: record = dict(zip(header, parts)) records.append(record) print(f' Record: {record}') return records ">>), {ok, Records} = py:eval(Ctx, <<"process_csv(buf)">>, #{<<"buf">> => Buf}), io:format("Parsed ~p records~n~n", [length(Records)]), ok. asyncio_demo() -> io:format("--- Async I/O Demo (Erlang streaming to Python) ---~n~n"), %% Create buffer that Erlang will fill {ok, Buf} = py_buffer:new(), Self = self(), Ctx = py:context(1), ok = py:exec(Ctx, <<" import asyncio async def async_buffer_reader(buf): '''Read from buffer asynchronously as Erlang streams data.''' chunks = [] read_count = 0 while not buf.at_eof(): available = buf.readable_amount() if available > 0: chunk = buf.read_nonblock(available) chunks.append(chunk) read_count += 1 print(f' [Python] Read chunk {read_count}: {len(chunk)} bytes') else: # Yield to event loop while waiting for Erlang to write more await asyncio.sleep(0.005) return b''.join(chunks) def run_async_reader(buf): '''Run async reader.''' loop = asyncio.new_event_loop() try: return loop.run_until_complete(async_buffer_reader(buf)) finally: loop.close() ">>), %% Spawn a process to stream data from Erlang spawn_link(fun() -> Chunks = [ <<"Hello from Erlang!">>, <<" Streaming chunk 2.">>, <<" Final chunk 3.">> ], lists:foreach(fun(Chunk) -> timer:sleep(30), %% Simulate network delay io:format(" [Erlang] Writing: ~p~n", [Chunk]), ok = py_buffer:write(Buf, Chunk) end, Chunks), timer:sleep(10), io:format(" [Erlang] Closing buffer (EOF)~n"), ok = py_buffer:close(Buf), Self ! writer_done end), %% Python reads asynchronously while Erlang writes io:format(" Starting async read while Erlang streams...~n"), {ok, Result} = py:eval(Ctx, <<"run_async_reader(buf)">>, #{<<"buf">> => Buf}), %% Wait for writer receive writer_done -> ok after 2000 -> ok end, io:format(" Final result: ~p~n~n", [Result]), ok.