// NewStream creates a stream for given path, listening for file events and
// calling fn upon receiving any.
346839
func newStream(path string, fn streamFunc) *stream {
	return &stream{
		path: path,
		info: streamFuncs.add(fn),
	}
}
1020700
func NewSocketListener(socketPath string, handler ConnectionHandler) (*socketListener, error) {
	listener, err := sockets.Listen(socketPath)
	if err != nil {
		return nil, errors.Trace(err)
	}
	sListener := &socketListener{listener: listener, handler: handler}
	sListener.t.Go(sListener.loop)
	return sListener, nil
}
264327
func (f *FileSystem) stream(conn io.ReadWriteCloser) {
	defer metrics.MeasureSince([]string{"client", "file_system", "stream"}, time.Now())
	defer conn.Close()

	// Decode the arguments
	var req cstructs.FsStreamRequest
	decoder := codec.NewDecoder(conn, structs.MsgpackHandle)
	encoder := codec.NewEncoder(conn, structs.MsgpackHandle)

	if err := decoder.Decode(&req); err != nil {
		f.handleStreamResultError(err, helper.Int64ToPtr(500), encoder)
		return
	}

	// Check read permissions
	if aclObj, err := f.c.ResolveToken(req.QueryOptions.AuthToken); err != nil {
		f.handleStreamResultError(err, nil, encoder)
		return
	} else if aclObj != nil && !aclObj.AllowNsOp(req.Namespace, acl.NamespaceCapabilityReadFS) {
		f.handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
		return
	}

	// Validate the arguments
	if req.AllocID == "" {
		f.handleStreamResultError(allocIDNotPresentErr, helper.Int64ToPtr(400), encoder)
		return
	}
	if req.Path == "" {
		f.handleStreamResultError(pathNotPresentErr, helper.Int64ToPtr(400), encoder)
		return
	}
	switch req.Origin {
	case "start", "end":
	case "":
		req.Origin = "start"
	default:
		f.handleStreamResultError(invalidOrigin, helper.Int64ToPtr(400), encoder)
		return
	}

	fs, err := f.c.GetAllocFS(req.AllocID)
	if err != nil {
		code := helper.Int64ToPtr(500)
		if structs.IsErrUnknownAllocation(err) {
			code = helper.Int64ToPtr(404)
		}

		f.handleStreamResultError(err, code, encoder)
		return
	}

	// Calculate the offset
	fileInfo, err := fs.Stat(req.Path)
	if err != nil {
		f.handleStreamResultError(err, helper.Int64ToPtr(400), encoder)
		return
	}
	if fileInfo.IsDir {
		f.handleStreamResultError(
			fmt.Errorf("file %q is a directory", req.Path),
			helper.Int64ToPtr(400), encoder)
		return
	}

	// If offsetting from the end subtract from the size
	if req.Origin == "end" {
		req.Offset = fileInfo.Size - req.Offset
		if req.Offset < 0 {
			req.Offset = 0
		}
	}

	frames := make(chan *sframer.StreamFrame, streamFramesBuffer)
	errCh := make(chan error)
	var buf bytes.Buffer
	frameCodec := codec.NewEncoder(&buf, structs.JsonHandle)

	// Create the framer
	framer := sframer.NewStreamFramer(frames, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
	framer.Run()
	defer framer.Destroy()

	// If we aren't following end as soon as we hit EOF
	var eofCancelCh chan error
	if !req.Follow {
		eofCancelCh = make(chan error)
		close(eofCancelCh)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Start streaming
	go func() {
		if err := f.streamFile(ctx, req.Offset, req.Path, req.Limit, fs, framer, eofCancelCh); err != nil {
			select {
			case errCh <- err:
			case <-ctx.Done():
			}
		}

		framer.Destroy()
	}()

	// Create a goroutine to detect the remote side closing
	go func() {
		for {
			if _, err := conn.Read(nil); err != nil {
				if err == io.EOF || err == io.ErrClosedPipe {
					// One end of the pipe was explicitly closed, exit cleanly
					cancel()
					return
				}
				select {
				case errCh <- err:
				case <-ctx.Done():
					return
				}
			}
		}
	}()

	var streamErr error
OUTER:
	for {
		select {
		case streamErr = <-errCh:
			break OUTER
		case frame, ok := <-frames:
			if !ok {
				// frame may have been closed when an error
				// occurred. Check once more for an error.
				select {
				case streamErr = <-errCh:
					// There was a pending error!
				default:
					// No error, continue on
				}

				break OUTER
			}

			var resp cstructs.StreamErrWrapper
			if req.PlainText {
				resp.Payload = frame.Data
			} else {
				if err = frameCodec.Encode(frame); err != nil {
					streamErr = err
					break OUTER
				}

				resp.Payload = buf.Bytes()
				buf.Reset()
			}

			if err := encoder.Encode(resp); err != nil {
				streamErr = err
				break OUTER
			}
			encoder.Reset(conn)
		case <-ctx.Done():
			break OUTER
		}
	}

	if streamErr != nil {
		f.handleStreamResultError(streamErr, helper.Int64ToPtr(500), encoder)
		return
	}
}
1154881
func (s *Swarm) NewStreamWithNetConn(netConn net.Conn) (*Stream, error) {
	c, err := s.AddConn(netConn)
	if err != nil {
		return nil, err
	}
	return s.NewStreamWithConn(c)
}
855434
func newStreamConfig(info logger.Info) (*logStreamConfig, error) {
	logGroupName := info.Config[logGroupKey]
	logStreamName, err := loggerutils.ParseLogTag(info, "{{.FullID}}")
	if err != nil {
		return nil, err
	}
	logCreateGroup := false
	if info.Config[logCreateGroupKey] != "" {
		logCreateGroup, err = strconv.ParseBool(info.Config[logCreateGroupKey])
		if err != nil {
			return nil, err
		}
	}

	logNonBlocking := info.Config["mode"] == "non-blocking"

	forceFlushInterval := defaultForceFlushInterval
	if info.Config[forceFlushIntervalKey] != "" {
		forceFlushIntervalAsInt, err := strconv.Atoi(info.Config[forceFlushIntervalKey])
		if err != nil {
			return nil, err
		}
		forceFlushInterval = time.Duration(forceFlushIntervalAsInt) * time.Second
	}

	maxBufferedEvents := int(defaultMaxBufferedEvents)
	if info.Config[maxBufferedEventsKey] != "" {
		maxBufferedEvents, err = strconv.Atoi(info.Config[maxBufferedEventsKey])
		if err != nil {
			return nil, err
		}
	}

	if info.Config[logStreamKey] != "" {
		logStreamName = info.Config[logStreamKey]
	}

	multilinePattern, err := parseMultilineOptions(info)
	if err != nil {
		return nil, err
	}

	containerStreamConfig := &logStreamConfig{
		logStreamName:      logStreamName,
		logGroupName:       logGroupName,
		logCreateGroup:     logCreateGroup,
		logNonBlocking:     logNonBlocking,
		forceFlushInterval: forceFlushInterval,
		maxBufferedEvents:  maxBufferedEvents,
		multilinePattern:   multilinePattern,
	}

	return containerStreamConfig, nil
}
643129
func (rc *RemoteContext) NewStream(timeout time.Duration) (*Stream, error) {
	if err := rc.ProbeConn(rc.conn); err != nil {
		return nil, err
	}

	ctx, cancel := context.WithCancel(context.TODO())
	stream, err := rc.Client.Step(ctx)
	if err != nil {
		cancel()
		return nil, errors.WithStack(err)
	}

	streamID := atomic.AddUint64(&rc.nextStreamID, 1)
	nodeName := commonNameFromContext(stream.Context())

	var canceled uint32

	abortChan := make(chan struct{})

	abort := func() {
		cancel()
		rc.streamsByID.Delete(streamID)
		rc.Metrics.reportEgressStreamCount(rc.Channel, atomic.LoadUint32(&rc.streamsByID.size))
		rc.Logger.Debugf("Stream %d to %s(%s) is aborted", streamID, nodeName, rc.endpoint)
		atomic.StoreUint32(&canceled, 1)
		close(abortChan)
	}

	once := &sync.Once{}
	abortReason := &atomic.Value{}
	cancelWithReason := func(err error) {
		abortReason.Store(err.Error())
		once.Do(abort)
	}

	logger := flogging.MustGetLogger("orderer.common.cluster.step")
	stepLogger := logger.WithOptions(zap.AddCallerSkip(1))

	s := &Stream{
		Channel:            rc.Channel,
		metrics:            rc.Metrics,
		abortReason:        abortReason,
		abortChan:          abortChan,
		sendBuff:           make(chan *orderer.StepRequest, rc.SendBuffSize),
		commShutdown:       rc.shutdownSignal,
		NodeName:           nodeName,
		Logger:             stepLogger,
		ID:                 streamID,
		Endpoint:           rc.endpoint,
		Timeout:            timeout,
		Cluster_StepClient: stream,
		Cancel:             cancelWithReason,
		canceled:           &canceled,
	}

	s.expCheck = &certificateExpirationCheck{
		minimumExpirationWarningInterval: rc.minimumExpirationWarningInterval,
		expirationWarningThreshold:       rc.certExpWarningThreshold,
		expiresAt:                        rc.expiresAt,
		endpoint:                         s.Endpoint,
		nodeName:                         s.NodeName,
		alert: func(template string, args ...interface{}) {
			s.Logger.Warningf(template, args...)
		},
	}

	rc.Logger.Debugf("Created new stream to %s with ID of %d and buffer size of %d",
		rc.endpoint, streamID, cap(s.sendBuff))

	rc.streamsByID.Store(streamID, s)
	rc.Metrics.reportEgressStreamCount(rc.Channel, atomic.LoadUint32(&rc.streamsByID.size))

	go func() {
		rc.workerCountReporter.increment(s.metrics)
		s.serviceStream()
		rc.workerCountReporter.decrement(s.metrics)
	}()

	return s, nil
}
1013547
func (m *Streamer) Stream(stdout, stderr chan []byte) StreamID {
	sid := m.idGen.next()
	m.streams.Store(sid, newStream(stdout, stderr))

	return sid
}
377414
func NewStochOscForStream(priceStream gotrade.DOHLCVStreamSubscriber, fastKTimePeriod int, slowKTimePeriod int, slowDTimePeriod int) (indicator *StochOsc, err error) {
	ind, err := NewStochOsc(fastKTimePeriod, slowKTimePeriod, slowDTimePeriod)
	priceStream.AddTickSubscription(ind)
	return ind, err
}
246762
func (pn *peernet) NewStream(ctx context.Context, p peer.ID) (inet.Stream, error) {
	c, err := pn.DialPeer(ctx, p)
	if err != nil {
		return nil, err
	}
	return c.NewStream()
}
958145
func NewStreamClient(params ClientParams) StreamClient {
	return newClient(params.ServiceName, params.ClientConfig, params.Options...)
}
314348
func NewStreamWatcher(d Decoder) *StreamWatcher {
	sw := &StreamWatcher{
		source: d,
		// It's easy for a consumer to add buffering via an extra
		// goroutine/channel, but impossible for them to remove it,
		// so nonbuffered is better.
		result: make(chan Event),
	}
	go sw.receive()
	return sw
}
222369
func NewStream(dataShards, parityShards int, o ...Option) (StreamEncoder, error) {
	enc, err := New(dataShards, parityShards, o...)
	if err != nil {
		return nil, err
	}
	rs := enc.(*reedSolomon)
	r := rsStream{r: rs, bs: 4 << 20}
	r.readShards = readShards
	r.writeShards = writeShards
	return &r, err
}
341583
func (c *Client) Stream(method, path string, in, out interface{}) (stream.Stream, error) {
	return c.StreamWithHeader(method, path, make(http.Header), in, out)
}
643983
func NewSyncedEventStream(stream EventStream) SelectableEventStream {
	var (
		sync syncedEventStream

		in   <-chan Event
		subs <-chan EventWriter
		halt <-chan eventStreamHaltReq
	)

	closeChans := func() func() {
		var (
			inCh   = make(chan Event)
			subsCh = make(chan EventWriter)
			haltCh = make(chan eventStreamHaltReq)
		)

		sync.in, in = inCh, inCh
		sync.subs, subs = subsCh, subsCh
		sync.halt, halt = haltCh, haltCh

		return func() {
			close(inCh)
			close(subsCh)
			close(haltCh)
		}
	}()

	go func() {
		var haltReq eventStreamHaltReq

	communication:
		for {
			select {
			case e := <-in:
				stream.Write(e)

			case w := <-subs:
				stream.Subscribe(w)

			case haltReq = <-halt:
				break communication
			}
		}

		closeChans()
		haltReq.hasHalted <- stream
	}()

	return sync
}
363784
func NewSalsaStream(key []byte) (*SalsaStream, error) {
	hash := sha256.Sum256(key)
	state := make([]uint32, 16)

	state[1] = u8to32little(hash[:], 0)
	state[2] = u8to32little(hash[:], 4)
	state[3] = u8to32little(hash[:], 8)
	state[4] = u8to32little(hash[:], 12)
	state[11] = u8to32little(hash[:], 16)
	state[12] = u8to32little(hash[:], 20)
	state[13] = u8to32little(hash[:], 24)
	state[14] = u8to32little(hash[:], 28)
	state[0] = sigmaWords[0]
	state[5] = sigmaWords[1]
	state[10] = sigmaWords[2]
	state[15] = sigmaWords[3]

	state[6] = u8to32little(iv, 0)
	state[7] = u8to32little(iv, 4)
	state[8] = uint32(0)
	state[9] = uint32(0)

	s := SalsaStream{
		State:        state,
		blockUsed:    64, // Ensure a fresh block is generated, the first time bytes are needed
		currentBlock: make([]byte, 0),
	}
	return &s, nil
}
971066
func NewStream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
	return DefaultClient.Stream(ctx, request, opts...)
}
971146
func (s *Swarm) NewStreamSelectConn(selConn SelectConn) (*Stream, error) {
	if selConn == nil {
		return nil, errors.New("nil SelectConn")
	}

	conns := s.Conns()
	if len(conns) == 0 {
		return nil, ErrNoConnections
	}
	return s.newStreamSelectConn(selConn, conns)
}