// NewStream creates a new stream on any available connection to peer, dialing
// if necessary.
556803
func (s *Swarm) NewStream(ctx context.Context, p peer.ID) (inet.Stream, error) {
	log.Debugf("[%s] opening stream to peer [%s]", s.local, p)

	// Algorithm:
	// 1. Find the best connection, otherwise, dial.
	// 2. Try opening a stream.
	// 3. If the underlying connection is, in fact, closed, close the outer
	//    connection and try again. We do this in case we have a closed
	//    connection but don't notice it until we actually try to open a
	//    stream.
	//
	// Note: We only dial once.
	//
	// TODO: Try all connections even if we get an error opening a stream on
	// a non-closed connection.
	dials := 0
	for {
		c := s.bestConnToPeer(p)
		if c == nil {
			if nodial, _ := inet.GetNoDial(ctx); nodial {
				return nil, inet.ErrNoConn
			}

			if dials >= DialAttempts {
				return nil, errors.New("max dial attempts exceeded")
			}
			dials++

			var err error
			c, err = s.dialPeer(ctx, p)
			if err != nil {
				return nil, err
			}
		}
		s, err := c.NewStream()
		if err != nil {
			if c.conn.IsClosed() {
				continue
			}
			return nil, err
		}
		return s, nil
	}
}
556606
func NewStreamFramer(out chan<- *StreamFrame,
	heartbeatRate, batchWindow time.Duration, frameSize int) *StreamFramer {

	// Create the heartbeat and flush ticker
	heartbeat := time.NewTicker(heartbeatRate)
	flusher := time.NewTicker(batchWindow)

	return &StreamFramer{
		out:        out,
		frameSize:  frameSize,
		heartbeat:  heartbeat,
		flusher:    flusher,
		f:          new(StreamFrame),
		data:       bytes.NewBuffer(make([]byte, 0, 2*frameSize)),
		shutdownCh: make(chan struct{}),
		exitCh:     make(chan struct{}),
	}
}
734478
func newStream(id uint32, frameSize int, sess *Session) *Stream {
	s := new(Stream)
	s.id = id
	s.chReadEvent = make(chan struct{}, 1)
	s.frameSize = frameSize
	s.sess = sess
	s.die = make(chan struct{})
	return s
}
401658
func NewStreamFileBuilder(writer io.Writer) *StreamFileBuilder {
	return &StreamFileBuilder{
		zipWriter:          zip.NewWriter(writer),
		xlsxFile:           NewFile(),
		cellTypeToStyleIds: make(map[CellType]int),
		maxStyleId:         initMaxStyleId,
	}
}
110615
func NewPubSocket() (*PubSocket, error) {
	s, err := NewSocket(AF_SP, PUB)
	return &PubSocket{s}, err
}
323644
func NewStream(w io.Writer) *Stream {
	return &Stream{
		Writer: w,
		enc:    json.NewEncoder(w),
	}
}
204139
func NewStream(preAuth RequestStream, processor Processor, postAuth ResultStream) Stream {
	if processor == nil {
		processor = newMemoryProcessor()
	}

	if preAuth != nil {
		linkRequest(preAuth, processor)
	}

	if postAuth != nil {
		linkResult(processor, postAuth)
		linkResult(postAuth, terminator{})
	} else {
		linkResult(processor, terminator{})
	}

	if preAuth != nil {
		return newStreamHead(preAuth)
	}

	return newStreamHead(processor)
}
397745
func (stc *ScatterConn) UpdateStream(ctx context.Context, rs *srvtopo.ResolvedShard, timestamp int64, position string, callback func(*querypb.StreamEvent) error) error {
	return rs.QueryService.UpdateStream(ctx, rs.Target, position, timestamp, callback)
}
96107
func NewServerStream(s Stream, options ...ServerStreamOption) (*ServerStream, error) {
	if s == nil {
		return nil, yarpcerrors.InvalidArgumentErrorf("non-nil stream is required")
	}
	return &ServerStream{stream: s}, nil
}
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
}
1232216
func NewStreamsByType() map[OperationType]map[uint64]*Stream {
	m := make(map[OperationType]map[uint64]*Stream)
	m[ConsensusOperation] = make(map[uint64]*Stream)
	m[SubmitOperation] = make(map[uint64]*Stream)
	return m
}
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)
}
336744
func NewServer(addr string) (*Server, error) {
	s := Server{
		cmds:  map[string]Cmd{},
		peers: map[net.Conn]struct{}{},
	}

	l, err := net.Listen("tcp", addr)
	if err != nil {
		return nil, err
	}
	s.l = l

	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		s.serve(l)
	}()
	return &s, nil
}
971066
func NewStream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
	return DefaultClient.Stream(ctx, request, opts...)
}
509354
func (conn *FakeVTGateConn) UpdateStream(ctx context.Context, keyspace string, shard string, keyRange *topodatapb.KeyRange, tabletType topodatapb.TabletType, timestamp int64, event *querypb.EventToken) (vtgateconn.UpdateStreamReader, error) {
	return nil, fmt.Errorf("NYI")
}
1251023
func newStream(streamID protocol.StreamID,
	sender streamSender,
	flowController flowcontrol.StreamFlowController,
	version protocol.VersionNumber,
) *stream {
	s := &stream{sender: sender, version: version}
	senderForSendStream := &uniStreamSender{
		streamSender: sender,
		onStreamCompletedImpl: func() {
			s.completedMutex.Lock()
			s.sendStreamCompleted = true
			s.checkIfCompleted()
			s.completedMutex.Unlock()
		},
	}
	s.sendStream = *newSendStream(streamID, senderForSendStream, flowController, version)
	senderForReceiveStream := &uniStreamSender{
		streamSender: sender,
		onStreamCompletedImpl: func() {
			s.completedMutex.Lock()
			s.receiveStreamCompleted = true
			s.checkIfCompleted()
			s.completedMutex.Unlock()
		},
	}
	s.receiveStream = *newReceiveStream(streamID, senderForReceiveStream, flowController, version)
	return s
}
20291
func (c *Client) NewStreamingRequest(path string, params interface{}) (*http.Request, error) {
	u, err := url.Parse(c.StreamBaseURL + path)
	if err != nil {
		return nil, err
	}

	return c.newRequest("GET", u, params, nil)
}