// StdStreams returns the standard streams (stdin, stdout, stderr).
351715
func StdStreams() (stdIn io.ReadCloser, stdOut, stdErr io.Writer) {
	return os.Stdin, os.Stdout, os.Stderr
}
649360
func (c *Config) CopyStreams(ctx context.Context, cfg *AttachConfig) <-chan error {
	var group errgroup.Group

	// Connect stdin of container to the attach stdin stream.
	if cfg.Stdin != nil {
		group.Go(func() error {
			logrus.Debug("attach: stdin: begin")
			defer logrus.Debug("attach: stdin: end")

			defer func() {
				if cfg.CloseStdin && !cfg.TTY {
					cfg.CStdin.Close()
				} else {
					// No matter what, when stdin is closed (io.Copy unblock), close stdout and stderr
					if cfg.CStdout != nil {
						cfg.CStdout.Close()
					}
					if cfg.CStderr != nil {
						cfg.CStderr.Close()
					}
				}
			}()

			var err error
			if cfg.TTY {
				_, err = copyEscapable(cfg.CStdin, cfg.Stdin, cfg.DetachKeys)
			} else {
				_, err = pools.Copy(cfg.CStdin, cfg.Stdin)
			}
			if err == io.ErrClosedPipe {
				err = nil
			}
			if err != nil {
				logrus.WithError(err).Debug("error on attach stdin")
				return errors.Wrap(err, "error on attach stdin")
			}
			return nil
		})
	}

	attachStream := func(name string, stream io.Writer, streamPipe io.ReadCloser) error {
		logrus.Debugf("attach: %s: begin", name)
		defer logrus.Debugf("attach: %s: end", name)
		defer func() {
			// Make sure stdin gets closed
			if cfg.Stdin != nil {
				cfg.Stdin.Close()
			}
			streamPipe.Close()
		}()

		_, err := pools.Copy(stream, streamPipe)
		if err == io.ErrClosedPipe {
			err = nil
		}
		if err != nil {
			logrus.WithError(err).Debugf("attach: %s", name)
			return errors.Wrapf(err, "error attaching %s stream", name)
		}
		return nil
	}

	if cfg.Stdout != nil {
		group.Go(func() error {
			return attachStream("stdout", cfg.Stdout, cfg.CStdout)
		})
	}
	if cfg.Stderr != nil {
		group.Go(func() error {
			return attachStream("stderr", cfg.Stderr, cfg.CStderr)
		})
	}

	errs := make(chan error, 1)
	go func() {
		defer logrus.Debug("attach done")
		groupErr := make(chan error, 1)
		go func() {
			groupErr <- group.Wait()
		}()
		select {
		case <-ctx.Done():
			// close all pipes
			if cfg.CStdin != nil {
				cfg.CStdin.Close()
			}
			if cfg.CStdout != nil {
				cfg.CStdout.Close()
			}
			if cfg.CStderr != nil {
				cfg.CStderr.Close()
			}

			// Now with these closed, wait should return.
			if err := group.Wait(); err != nil {
				errs <- err
				return
			}
			errs <- ctx.Err()
		case err := <-groupErr:
			errs <- err
		}
	}()

	return errs
}
547194
func (s *CloudWatchLogsConfiguration) SetLogStreams(v []*CloudWatchLogsLogStream) *CloudWatchLogsConfiguration {
	s.LogStreams = v
	return s
}
1030011
func (c *Container) initStreams() error {
	if c.streams != nil {
		return nil
	}
	if !c.p.IsAlive() {
		c.Log(ERROR, "can not init stream to a non-existing sandbox")
		return errors.ErrPodNotAlive.WithArgs(c.p.Id())
	}
	c.Log(TRACE, "init io streams")
	c.streams = NewStreamConfig()
	c.streams.NewInputPipes()
	tty := &hypervisor.TtyIO{
		Stdin:  c.streams.Stdin(),
		Stdout: c.streams.Stdout(),
		Stderr: c.streams.Stderr(),
	}
	if err := c.p.sandbox.Attach(tty, c.Id()); err != nil {
		c.Log(ERROR, err)
		return err
	}
	return nil
}
36237
func (c *Conn) NewStream() (inet.Stream, error) {
	ts, err := c.conn.OpenStream()
	if err != nil {
		return nil, err
	}
	return c.addStream(ts, inet.DirOutbound)
}
413600
func newStream(r io.Reader) *stream {
	return &stream{r, make([]byte, buffer), buffer}
}
339938
func DispatchStreamHandler(
	h StreamHandler,
	stream *ServerStream,
) (err error) {
	return InvokeStreamHandler(StreamInvokeRequest{
		Stream:  stream,
		Handler: h,
	})
}
954355
func newDiscMsgStream(p *peer) *msgStream {
	return newMsgStream(p,
		"Update stream for gossiper created",
		"Update stream for gossiper exited",
		1000,
		func(msg lnwire.Message) {
			p.server.authGossiper.ProcessRemoteAnnouncement(msg, p)
		},
	)
}
842349
func matchStreams(streams []*coordinator.LogStream, pattern string) []*coordinator.LogStream {
	// Filter results that match the requested stream name.
	// ** turns into .*
	// * turns into [^/]*
	doubles := strings.Split(pattern, "**")
	for i, d := range doubles {
		singles := strings.Split(d, "*")
		for j, s := range singles {
			singles[j] = regexp.QuoteMeta(s)
		}
		doubles[i] = strings.Join(singles, "[^/]*")
	}
	r := regexp.MustCompile(fmt.Sprintf("^%s$", strings.Join(doubles, ".*")))
	result := make([]*coordinator.LogStream, 0, len(streams))
	for _, stream := range streams {
		if stream.Purged {
			continue
		}
		if stream.StreamType != logpb.StreamType_TEXT {
			// Multistream viewing only supported for text streams.
			continue
		}
		if r.MatchString(stream.Name) {
			result = append(result, stream)
		}
	}
	return result
}
1118871
func (self *Transcoder) Streams() (streams []av.CodecData, err error) {
	for _, stream := range self.streams {
		streams = append(streams, stream.codec)
	}
	return
}
659262
func NewStdDevForStreamWithSrcLen(sourceLength uint, priceStream gotrade.DOHLCVStreamSubscriber, timePeriod int, selectData gotrade.DOHLCVDataSelectionFunc) (indicator *StdDev, err error) {
	ind, err := NewStdDevWithSrcLen(sourceLength, timePeriod, selectData)
	priceStream.AddTickSubscription(ind)
	return ind, err
}
696122
func (be *BlockExecution) StreamEvents() StreamEvents {
	var ses StreamEvents
	ses = append(ses, &StreamEvent{
		BeginBlock: &BeginBlock{
			Height: be.Height,
			Header: be.Header,
		},
	})
	for _, txe := range be.TxExecutions {
		ses = append(ses, txe.StreamEvents()...)
	}
	return append(ses, &StreamEvent{
		EndBlock: &EndBlock{
			Height: be.Height,
		},
	})
}
1164362
func (r *Runtime) ReadStandardStreams() (stdout []byte, stderr []byte, err error) {
	_, err = r.stdout.Seek(0, io.SeekStart)
	stdout, err2 := ioutil.ReadAll(r.stdout)
	if err == nil && err2 != nil {
		err = err2
	}
	_, err = r.stderr.Seek(0, io.SeekStart)
	stderr, err2 = ioutil.ReadAll(r.stderr)
	if err == nil && err2 != nil {
		err = err2
	}
	return stdout, stderr, err
}
881776
func WithStdio(opt *Streams) {
	WithStreams(os.Stdin, os.Stdout, os.Stderr)(opt)
}
338071
func (sbc *SandboxConn) VStreamRows(ctx context.Context, target *querypb.Target, query string, lastpk *querypb.QueryResult, send func(*binlogdatapb.VStreamRowsResponse) error) error {
	return fmt.Errorf("not implemented in test")
}
1056803
func (st largeBlobStreamer) StreamBlobs(ctx context.Context, dest chan<- blobserver.BlobAndToken, contToken string) (err error) {
	defer close(dest)
	s := st.sto
	large := s.large

	var after string // for enumerateAll
	var skipFiles int
	var firstRef blob.Ref // first we care about

	if contToken != "" {
		f := strings.SplitN(contToken, ":", 2)
		if len(f) != 2 {
			return errContToken
		}
		firstRef, _ = blob.Parse(f[0])
		skipFiles, err = strconv.Atoi(f[1])
		if !firstRef.Valid() || err != nil {
			return errContToken
		}
		// EnumerateAllFrom takes a cursor that's greater, but
		// we want to start _at_ firstRef. So start
		// enumerating right before our target.
		after = firstRef.StringMinusOne()
	}
	return blobserver.EnumerateAllFrom(ctx, large, after, func(sb blob.SizedRef) error {
		if firstRef.Valid() {
			if sb.Ref.Less(firstRef) {
				// Skip.
				return nil
			}
			if firstRef.Less(sb.Ref) {
				skipFiles = 0 // reset it.
			}
		}
		fileN := 0
		return s.foreachZipBlob(ctx, sb.Ref, func(bap BlobAndPos) error {
			if skipFiles > 0 {
				skipFiles--
				fileN++
				return nil
			}
			select {
			case dest <- blobserver.BlobAndToken{
				Blob: blob.NewBlob(bap.Ref, bap.Size, func(ctx context.Context) ([]byte, error) {
					return slurpSizedRef(ctx, s, bap.SizedRef)
				}),
				Token: fmt.Sprintf("%s:%d", sb.Ref, fileN),
			}:
				fileN++
				return nil
			case <-ctx.Done():
				return ctx.Err()
			}
		})
	})
}
510370
func (q *query) VStream(request *binlogdatapb.VStreamRequest, stream queryservicepb.Query_VStreamServer) (err error) {
	defer q.server.HandlePanic(&err)
	ctx := callerid.NewContext(callinfo.GRPCCallInfo(stream.Context()),
		request.EffectiveCallerId,
		request.ImmediateCallerId,
	)
	err = q.server.VStream(ctx, request.Target, request.Position, request.Filter, func(events []*binlogdatapb.VEvent) error {
		return stream.Send(&binlogdatapb.VStreamResponse{
			Events: events,
		})
	})
	return vterrors.ToGRPC(err)
}