Browse Source

增加调试代码

gitlink
Sydonian 1 year ago
parent
commit
32254423e9
2 changed files with 9 additions and 2 deletions
  1. +5
    -1
      agent/internal/grpc/io.go
  2. +4
    -1
      common/pkgs/ioswitch2/ops2/ipfs.go

+ 5
- 1
agent/internal/grpc/io.go View File

@@ -60,10 +60,14 @@ func (s *Service) SendStream(server agtrpc.Agent_SendStreamServer) error {


pr, pw := io.Pipe() pr, pw := io.Pipe()


rb := io2.RingBuffer(pr, 16*1024)
rb.UpstreamName = fmt.Sprintf("GRPC(send) input")
rb.DownstreamName = fmt.Sprintf("GRPC(send) output %v", msg.VarID)

varID := exec.VarID(msg.VarID) varID := exec.VarID(msg.VarID)
sw.PutVars(&exec.StreamVar{ sw.PutVars(&exec.StreamVar{
ID: varID, ID: varID,
Stream: pr,
Stream: rb,
}) })


// 然后读取文件数据 // 然后读取文件数据


+ 4
- 1
common/pkgs/ioswitch2/ops2/ipfs.go View File

@@ -44,7 +44,10 @@ func (o *IPFSRead) Execute(ctx context.Context, e *exec.Executor) error {
defer file.Close() defer file.Close()


fut := future.NewSetVoid() fut := future.NewSetVoid()
o.Output.Stream = io2.AfterReadClosedOnce(file, func(closer io.ReadCloser) {
rb := io2.RingBuffer(file, 16*1024)
rb.UpstreamName = "IPFS"
rb.DownstreamName = fmt.Sprintf("IPFS output %v", o.Output.ID)
o.Output.Stream = io2.AfterReadClosedOnce(rb, func(closer io.ReadCloser) {
fut.SetVoid() fut.SetVoid()
}) })
e.PutVars(o.Output) e.PutVars(o.Output)


Loading…
Cancel
Save