runtime.LockOSThread()
defer runtime.UnlockOSThread()
handle := {{ start_call }}
if handle == 0 {
if err := lastError(); err != nil {
return nil, err
}
return nil, fmt.Errorf("failed to start {{ method_name }} stream")
}
ch := make(chan {{ item_type }})
stream := &{{ stream_type_name }}{ch: ch}
go func() {
defer close(ch)
defer C.{{ fn_free }}(handle)
for {
runtime.LockOSThread()
chunkPtr := C.{{ fn_next }}(handle)
if chunkPtr == 0 {
// Null means either a clean end-of-stream (errno 0) or a stream error
// (errno != 0); read lastError() while still locked to this OS thread,
// since the native last-error slot is per-OS-thread.
stream.err = lastError()
runtime.UnlockOSThread()
return
}
jsonPtr := C.{{ item_to_json_fn }}(chunkPtr)
if jsonPtr == nil {
C.{{ item_free_fn }}(chunkPtr)
stream.err = fmt.Errorf("{{ method_name }}: stream item could not be converted to JSON")
runtime.UnlockOSThread()
return
}
{% if item_is_sum_type %}
chunk, unmarshalErr := Unmarshal{{ item_type }}([]byte(C.GoString(jsonPtr)))
{% else %}
var chunk {{ item_type }}
unmarshalErr := json.Unmarshal([]byte(C.GoString(jsonPtr)), &chunk)
{% endif %}
C.{{ free_string_fn }}(jsonPtr)
C.{{ item_free_fn }}(chunkPtr)
runtime.UnlockOSThread()
if unmarshalErr != nil {
stream.err = fmt.Errorf("{{ method_name }}: failed to decode stream item: %w", unmarshalErr)
return
}
// Issue #448: guard the send against the caller cancelling ctx, so a consumer that
// stops ranging Chan() early (e.g. via break) does not park this goroutine on an
// unbuffered channel forever -- that used to leak both the goroutine and the native
// stream handle, since the deferred close(ch)/C.{{ fn_free }}(handle) above never
// ran. Deliberately NOT also checking ctx.Done() at the top of this loop, before
// calling C.{{ fn_next }}: that would skip one wasted native fetch after
// cancellation, but trades it for scheduling nondeterminism with no correctness
// benefit, since this send already guarantees the goroutine exits promptly either
// way. ~keep
select {
case ch <- chunk:
case <-ctx.Done():
stream.err = ctx.Err()
return
}
}
}()
return stream, nil
}