chore: modify stream copy

* chore: remove stream reader close protection

* chore: genericTransformWithCallbacks stream copy

See merge request: !219

Author: N3kox

compose/utils.go

@@ -206,6 +206,7 @@ func genericTransformWithCallbacks(t transform) transform {
 		}
 
 		ctx, is = callbacks.OnStartWithStreamInput(ctx, is)
+		is.Close() // goroutine free copy buffer release
 
 		output, err = t(ctx, inArr[0], opts...)
 		if err != nil {
@@ -219,7 +220,8 @@ func genericTransformWithCallbacks(t transform) transform {
 			return outArr[0], nil
 		}
 
-		_, _ = callbacks.OnEndWithStreamOutput(ctx, os)
+		_, os = callbacks.OnEndWithStreamOutput(ctx, os)
+		os.Close()
 
 		return outArr[0], nil
 	}

schema/stream.go

@@ -23,7 +23,6 @@ import (
 	"reflect"
 	"runtime/debug"
 	"sync"
-	"sync/atomic"
 
 	"github.com/cloudwego/eino/utils/safe"
 )
@@ -305,10 +304,6 @@ func (s *stream[T]) closeSend() {
 }
 
 func (s *stream[T]) closeRecv() {
-	if !atomic.CompareAndSwapUint32(&s.isClosed, 0, 1) {
-		return
-	}
-
 	close(s.closed)
 }