Function NewBinaryProcessor

core/diff/stream_windows.go:40–111

NewBinaryProcessor returns a binary processor for use with processing content streams

(ctx context.Context, imt, rmt string, stream StreamProcessor, name string, args, env []string, payload typeurl.Any)

// NewBinaryProcessor returns a binary processor for use with processing content streams
func NewBinaryProcessor(ctx context.Context, imt, rmt string, stream StreamProcessor, name string, args, env []string, payload typeurl.Any) (StreamProcessor, error) {
	cmd := exec.CommandContext(ctx, name, args...)
	cmd.Env = os.Environ()
	cmd.Env = append(cmd.Env, env...)

	if payload != nil {
		pb := typeurl.MarshalProto(payload)
		data, err := proto.Marshal(pb)
		if err != nil {
			return nil, err
		}
		up, err := getUiqPath()
		if err != nil {
			return nil, err
		}
		path := fmt.Sprintf("\\\\.\\pipe\\containerd-processor-%s-pipe", up)
		l, err := winio.ListenPipe(path, nil)
		if err != nil {
			return nil, err
		}
		go func() {
			defer l.Close()
			conn, err := l.Accept()
			if err != nil {
				log.G(ctx).WithError(err).Error("accept npipe connection")
				return
			}
			io.Copy(conn, bytes.NewReader(data))
			conn.Close()
		}()
		cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", processorPipe, path))
	}
	cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", mediaTypeEnvVar, imt))
	var (
		stdin  io.Reader
		closer func() error
		err    error
	)
	if f, ok := stream.(RawProcessor); ok {
		stdin = f.File()
		closer = f.File().Close
	} else {
		stdin = stream
	}
	cmd.Stdin = stdin
	r, w, err := os.Pipe()
	if err != nil {
		return nil, err
	}
	cmd.Stdout = w
	stderr := bytes.NewBuffer(nil)
	cmd.Stderr = stderr

	if err := cmd.Start(); err != nil {
		return nil, err
	}
	p := &binaryProcessor{
		cmd:    cmd,