code: func (s *scheduler) newRequestWithFunc(e *edge, f func(context.Context) (interface{}, error)) pipe.Receiver {
  	pp, start := pipe.NewWithFunction(f)
  	p := &edgePipe{
  		Pipe: pp,
  		From: e,
  	}
  	p.OnSendCompletion = func() {
  		p.mu.Lock()
  		defer p.mu.Unlock()
  		s.signal(p.From)
  	}
  	s.outgoing[e] = append(s.outgoing[e], p)
  	go start()
  	return p.Receiver
  }
docs: // newRequestWithFunc creates a new request pipe that invokes a async function