// Output concurrently collects outputs from previous step to the driver.
764079
func (d *Dataset) Output(f func(io.Reader) error) *Dataset {
	step := d.Flow.AddAllToOneStep(d, nil)
	step.IsOnDriverSide = true
	step.Name = "Output"
	step.Function = func(readers []io.Reader, writers []io.Writer, stat *pb.InstructionStat) error {
		errChan := make(chan error, len(readers))
		for i, reader := range readers {
			go func(i int, reader io.Reader) {
				errChan <- f(reader)
			}(i, reader)
		}
		for range readers {
			err := <-errChan
			if err != nil {
				fmt.Fprintf(os.Stderr, "Failed to process output: %v\n", err)
				return err
			}
		}
		return nil
	}
	return d
}
716650
func (oc *OutputChannel) Flush() {
	for {
		select {
		case msg := <-oc.input:
			if err := oc.output.Write(msg); err != nil {
				critialLogger.Print("cannot write log message: ", err.Error())
			}
		default:
			return
		}
	}
}
529510
func (cmd *CLI) StdOut() io.Writer {
	if cmd.stdOutput == nil {
		cmd.stdOutput = os.Stdout
	}
	return cmd.stdOutput
}
608203
func (fo *FileWriterOutput) Output(ll LogLevel, message []byte) {
	if fo.WriterOutput == nil {
		fh, err := fo.openFile()
		if err != nil {
			fo.fallbackLog("Could not open %#v: %s", fo.path, err)
			return
		}
		fo.WriterOutput = NewWriterOutput(fh)
	}
	fo.WriterOutput.Output(ll, message)
}
1172770
func (p Paragraph) Output(out io.Writer) {
	s := []byte(spacify(string(p.data)))
	if len(s) == 0 {
		out.Write(nl)
		return
	}
	indent := p.indent * INDENT
	width := p.cols - indent - len(p.prefix)
	wrapped := text.WrapBytes(s, width)
	lines := bytes.Split(wrapped, nl)
	gutter := makePad(indent)
	pad := makePad(len(p.prefix))

	for i, line := range lines {
		out.Write(gutter)
		if i == 0 {
			out.Write([]byte(p.prefix))
		} else {
			out.Write(pad)
		}
		out.Write(line)
		out.Write(nl)
	}
}
978985
func (c *C) Output(calldepth int, s string) error {
	d := time.Now().Sub(c.startTime)
	msec := d / time.Millisecond
	sec := d / time.Second
	min := d / time.Minute

	c.Logf("[LOG] %d:%02d.%03d %s", min, sec%60, msec%1000, s)
	return nil
}
424457
func (o *Output) Close() error {
	o.Lock()
	defer o.Unlock()
	var firstErr error
	for _, dst := range o.dests {
		if closer, ok := dst.(io.Closer); ok {
			err := closer.Close()
			if err != nil && firstErr == nil {
				firstErr = err
			}
		}
	}
	o.tasks.Wait()
	o.dests = nil
	return firstErr
}
556891
func (res NodeModuleResult) Output() (out string) {
	if res.Error != nil {
		out = res.Error.Error()
	} else {
		out = res.Message
	}
	return
}
1307395
func (p *Printer) AddOutput(writers ...io.Writer) *Printer {
	p.mu.Lock()
	defer p.mu.Unlock()

	for _, w := range writers {
		// set is terminal to false
		// if at least one of the writers
		// is not a terminal-based.
		if !terminal.IsTerminal(w) {
			p.IsTerminal = false
			break
		}
	}

	w := io.MultiWriter(append(writers, p.Output)...)
	p.Output = w
	return p

	// p.mu.Lock()
	// oldW := p.Output
	// newW := io.MultiWriter(writers...)
	// p.Output = writerFunc(func(p []byte) (n int, err error) {
	// 	n, err = oldW.Write(p)
	// 	if err != nil {
	// 		return
	// 	}
	// 	if n != len(p) {
	// 		err = io.ErrShortWrite
	// 		return
	// 	}
	// 	return newW.Write(p)
	// })
	// p.mu.Unlock()
}
1318001
func (d *pcXTDriver) Flush() error {
	defer func() {
		d.buffer = nil
	}()
	sc, err := chunkScanCodes(d.buffer, d.scancodeChunkSize)
	if err != nil {
		return err
	}
	for _, b := range sc {
		if err := d.sendImpl(b); err != nil {
			return err
		}
		time.Sleep(d.interval)
	}
	return nil
}
1105770
func (l Logger) Output(w io.Writer) Logger {
	l2 := New(w)
	l2.level = l.level
	l2.sampler = l.sampler
	if len(l.hooks) > 0 {
		l2.hooks = append(l2.hooks, l.hooks...)
	}
	if l.context != nil {
		l2.context = make([]byte, len(l.context), cap(l.context))
		copy(l2.context, l.context)
	}
	return l2
}
8637
func (o Output) Files() []string {
	f := make([]string, 0, len(o.Data))
	for k := range o.Data {
		f = append(f, k)
	}
	sort.Strings(f)
	return f
}
337699
func (cfg *cloudConfig) Output(kind OutputKind) (stdout, stderr string) {
	if out, ok := cfg.attrs["output"].(map[string]interface{}); ok {
		switch out := out[string(kind)].(type) {
		case string:
			stdout = out
		case []string:
			stdout, stderr = out[0], out[1]
		}
	}

	return stdout, stderr
}
105117
func (e *ExecRunner) CombinedOutput(cmd string) (string, error) {
	var b bytes.Buffer
	err := e.CombinedOutputTo(cmd, &b)
	if err != nil {
		return "", errors.Wrapf(err, "running command: %s\n output: %s", cmd, b.Bytes())
	}
	return b.String(), nil

}
220942
func Out(p []byte) *Stepper {
	return &Stepper{OPS, SPO, SPO, Triple{nil, p, nil, nil}, nil, make([]func(Path), 0, 0), nil}
}
1356406
func Output(environment, name string) (string, error) {
	cmd := exec.Command("sh", "-c", fmt.Sprintf("terraform output %s", name))
	cmd.Dir = filepath.Join(Dir, environment)

	out, err := cmd.CombinedOutput()
	if err != nil {
		return "", err
	}

	return strings.Trim(string(out), "\n"), nil
}
1360021
func copyOutput(r io.Reader, doneCh chan<- struct{}) {
	defer close(doneCh)

	pr, err := prefixedio.NewReader(r)
	if err != nil {
		panic(err)
	}

	stderrR, err := pr.Prefix(ErrorPrefix)
	if err != nil {
		panic(err)
	}
	stdoutR, err := pr.Prefix(OutputPrefix)
	if err != nil {
		panic(err)
	}
	defaultR, err := pr.Prefix("")
	if err != nil {
		panic(err)
	}

	var wg sync.WaitGroup
	wg.Add(3)
	go func() {
		defer wg.Done()
		io.Copy(os.Stderr, stderrR)
	}()
	go func() {
		defer wg.Done()
		io.Copy(os.Stdout, stdoutR)
	}()
	go func() {
		defer wg.Done()
		io.Copy(os.Stdout, defaultR)
	}()

	wg.Wait()
}