// Continuously pulls connections from the supplied Listener and returns them as
// connections to streamParamsC for consumption by Next().
420113
func (s *listenerStreamServer) serve() {
	defer close(s.acceptFinishedC)

	nextID := 0
	clientWG := sync.WaitGroup{}
	for {
		log.Debugf(s, "Beginning Accept() loop cycle.")
		conn, err := s.l.Accept()
		if err != nil {
			if atomic.LoadInt32(&s.closed) != 0 {
				log.WithError(err).Debugf(s, "Error during Accept() encountered (closed).")
			} else {
				log.WithError(err).Errorf(s, "Error during Accept().")
			}
			break
		}

		// Spawn a goroutine to handle this connection. This goroutine will take
		// ownership of the connection, closing it as appropriate.
		client := &streamClient{
			closedC: s.closedC,
			id:      nextID,
			conn:    &iotools.DeadlineReader{conn, 0},
		}
		client.Context = log.SetFields(s, log.Fields{
			"id":     client.id,
			"local":  client.conn.LocalAddr(),
			"remote": client.conn.RemoteAddr(),
		})

		clientWG.Add(1)
		go func() {
			defer clientWG.Done()

			var params *streamParams
			paniccatcher.Do(func() {
				var err error
				params, err = client.handle()
				if err != nil {
					log.Fields{
						log.ErrorKey: err,
					}.Errorf(client, "Failed to negotitate stream client.")
					params = nil
				}
			}, func(p *paniccatcher.Panic) {
				log.Fields{
					"panicReason": p.Reason,
				}.Errorf(client, "Panic during client handshake:\n%s", p.Stack)
				params = nil
			})

			// Did the negotiation fail?
			if params != nil {
				// Punt the client to our Next function. If we close while waiting, close
				// the Client.
				select {
				case s.streamParamsC <- params:
					break

				case <-s.closedC:
					log.Warningf(client, "Closed with client in hand. Cleaning up client.")
					params.rc.Close()
					params = nil
				}
			}

			// (Testing) write failed connections to discardC if configured.
			if params == nil && s.discardC != nil {
				s.discardC <- client
			}
		}()
		nextID++
	}

	// Wait for client connections to finish.
	clientWG.Wait()
	log.Fields{
		"totalConnections": nextID,
	}.Infof(s, "Exiting serve loop.")
}
1321812
func (a *AuthServer) GetTunnelConnections(clusterName string, opts ...services.MarshalOption) ([]services.TunnelConnection, error) {
	return a.GetCache().GetTunnelConnections(clusterName, opts...)
}
582292
func fetch(c context.Context, ch chan<- logResp, allParams []fetchParams) {
	defer close(ch) // Close the channel when we're done fetching.

	// TODO(hinoka): Remove this check once multistream view is supported.
	if len(allParams) > 1 {
		streamPaths := make([]string, len(allParams))
		for i, p := range allParams {
			streamPaths[i] = string(p.stream.Path())
			p.storage.Close()
		}
		ch <- logResp{nil, nil, fmt.Errorf("Only single stream supported, found %d: %s", len(allParams), strings.Join(streamPaths, "\n"))}
		return
	}

	// Extract the params we need.
	params := allParams[0]
	state := params.state
	st := params.storage
	defer st.Close() // Close the connection to the backend when we're done.

	index := types.MessageIndex(0)
	backoff := time.Second // How long to wait between fetch requests from storage.
	var err error
	for {
		// Get the terminal index from the LogStreamState again if we need to.
		// This is useful when the log stream terminated after the user request started.
		if !state.Terminated() {
			if err = datastore.Get(c, state); err != nil {
				ch <- logResp{nil, nil, err}
				return
			}
		}

		// Do the actual work.
		nextIndex, err := fetchOnce(c, ch, index, params)
		// Signal regardless of error.  Server will bail on error and flush otherwise.
		// Important: if fetchOnce errors out, we still expect nextIndex to increment regardless,
		// otherwise this may get stuck in an infinite loop.
		ch <- logResp{nil, nil, err}

		// Check if the log has finished streaming and if we're done.
		if state.Terminated() && nextIndex > types.MessageIndex(state.TerminalIndex) {
			logging.Fields{
				"terminalIndex": state.TerminalIndex,
				"currentIndex":  index,
			}.Debugf(c, "Finished streaming")
			return
		}

		if err != nil {
			index = nextIndex
			continue
		}

		// Log is still streaming.  Set the next index, sleep a bit and try again.
		backoff = backoff * 2
		if backoff > maxBackoff {
			backoff = maxBackoff
		}
		if index != nextIndex {
			if err != nil {
				// An error on one log entry means we want to try the next one asap.
				backoff = 0
			} else {
				// If its a relatively active log stream, don't sleep too long.
				backoff = time.Second
			}
		}
		if tr := clock.Sleep(c, backoff); tr.Err != nil {
			// If the user cancelled the request, bail out instead.
			ch <- logResp{nil, nil, tr.Err}
			return
		}
		index = nextIndex
	}
}
490399
func (h *portForwardProxy) getStreamPair(requestID string) (*httpStreamPair, bool) {
	h.streamPairsLock.Lock()
	defer h.streamPairsLock.Unlock()

	if p, ok := h.streamPairs[requestID]; ok {
		log.Infof("(conn=%p, request=%s) found existing stream pair", h.sourceConn, requestID)
		return p, false
	}

	log.Infof("(conn=%p, request=%s) creating new stream pair", h.sourceConn, requestID)

	p := newPortForwardPair(requestID)
	h.streamPairs[requestID] = p

	return p, true
}
588378
func (e *streamExecutor) Dial(protocols ...string) (httpstream.Connection, string, error) {
	rt := transport.DebugWrappers(e.transport)

	// TODO the client probably shouldn't be created here, as it doesn't allow
	// flexibility to allow callers to configure it.
	client := &http.Client{Transport: rt}

	req, err := http.NewRequest(e.method, e.url.String(), nil)
	if err != nil {
		return nil, "", fmt.Errorf("error creating request: %v", err)
	}
	for i := range protocols {
		req.Header.Add(httpstream.HeaderProtocolVersion, protocols[i])
	}

	resp, err := client.Do(req)
	if err != nil {
		return nil, "", fmt.Errorf("error sending request: %v", err)
	}
	defer resp.Body.Close()

	conn, err := e.upgrader.NewConnection(resp)
	if err != nil {
		return nil, "", err
	}

	return conn, resp.Header.Get(httpstream.HeaderProtocolVersion), nil
}
1177044
func (l *Listener) listenerConnLoop() {
	var nextReconnect time.Time

	reconnectInterval := l.minReconnectInterval
	for {
		for {
			err := l.connect()
			if err == nil {
				break
			}

			if l.closed() {
				return
			}
			l.emitEvent(ListenerEventConnectionAttemptFailed, err)

			time.Sleep(reconnectInterval)
			reconnectInterval *= 2
			if reconnectInterval > l.maxReconnectInterval {
				reconnectInterval = l.maxReconnectInterval
			}
		}

		if nextReconnect.IsZero() {
			l.emitEvent(ListenerEventConnected, nil)
		} else {
			l.emitEvent(ListenerEventReconnected, nil)
			l.Notify <- nil
		}

		reconnectInterval = l.minReconnectInterval
		nextReconnect = time.Now().Add(reconnectInterval)

		for {
			notification, ok := <-l.connNotificationChan
			if !ok {
				// lost connection, loop again
				break
			}
			l.Notify <- notification
		}

		err := l.disconnectCleanup()
		if l.closed() {
			return
		}
		l.emitEvent(ListenerEventDisconnected, err)

		time.Sleep(nextReconnect.Sub(time.Now()))
	}
}
927827
func (sc *SlaveConnection) streamEvents(ctx context.Context) chan mysql.BinlogEvent {
	// FIXME(alainjobart) I think we can use a buffered channel for better performance.
	eventChan := make(chan mysql.BinlogEvent)

	// Start reading events.
	sc.wg.Add(1)
	go func() {
		defer func() {
			close(eventChan)
			sc.wg.Done()
		}()
		for {
			event, err := sc.Conn.ReadBinlogEvent()
			if err != nil {
				if sqlErr, ok := err.(*mysql.SQLError); ok && sqlErr.Number() == mysql.CRServerLost {
					// CRServerLost = Lost connection to MySQL server during query
					// This is not necessarily an error. It could just be that we closed
					// the connection from outside.
					log.Infof("connection closed during binlog stream (possibly intentional): %v", err)
					return
				}
				log.Errorf("read error while streaming binlog events: %v", err)
				return
			}

			select {
			case eventChan <- event:
			case <-ctx.Done():
				return
			}
		}
	}()
	return eventChan
}
172796
func (d *dispatcher) Dispatch(g grim.GrimReaper, conn net.Conn) {

	// Create buffer and read connection type
	buf := make([]byte, 1)
	if _, err := conn.Read(buf); err != nil {
		d.logger.Warn("error reading StreamType",
			"remote", conn.RemoteAddr().String, "err", err)
		conn.Close()
		return
	}

	// Dispatch connection based on first byte
	ctype := StreamType(buf[0])
	if handler, ok := d.handlers[ctype]; ok {
		g.SpawnFunc(func(c context.Context) {
			handler.Handle(c, conn)
		})
		return
	}

	// Process unknown streams if there is an unknown stream handler
	if d.unknown != nil {
		g.SpawnFunc(func(c context.Context) {
			d.unknown.Handle(c, conn)
		})
		return
	}

	// If there is not an unknown stream handler, close the stream.
	d.logger.Warn("closing connection: unknown stream type",
		"type", ctype, "remote", conn.RemoteAddr().String())
	conn.Close()
}
1296961
func (c *MockConnection) Receive(eventch chan<- interface{}) {
	for {
		select {
		case e, ok := <-c.producerch:
			if !ok {
				return
			}
			eventch <- e
		case e, ok := <-c.rcvch:
			if !ok {
				return
			}
			eventch <- e
		}
	}
}
1169326
func NewStream(ctx context.Context, method string, conn *Conn) (Stream, error) {
	if ctx == nil {
		ctx = context.Background()
	}

	return newStreamClient(ctx, method, conn)
}
1224712
func (stc *ScatterConn) MessageStream(ctx context.Context, rss []*srvtopo.ResolvedShard, name string, callback func(*sqltypes.Result) error) error {
	// The cancelable context is used for handling errors
	// from individual streams.
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// mu is used to merge multiple callback calls into one.
	var mu sync.Mutex
	fieldSent := false
	lastErrors := newTimeTracker()
	allErrors := stc.multiGo(ctx, "MessageStream", rss, topodatapb.TabletType_MASTER, func(rs *srvtopo.ResolvedShard, i int) error {
		// This loop handles the case where a reparent happens, which can cause
		// an individual stream to end. If we don't succeed on the retries for
		// messageStreamGracePeriod, we abort and return an error.
		for {
			err := rs.QueryService.MessageStream(ctx, rs.Target, name, func(qr *sqltypes.Result) error {
				lastErrors.Reset(rs.Target)
				return stc.processOneStreamingResult(&mu, &fieldSent, qr, callback)
			})
			// nil and EOF are equivalent. UNAVAILABLE can be returned by vttablet if it's demoted
			// from master to replica. For any of these conditions, we have to retry.
			if err != nil && err != io.EOF && vterrors.Code(err) != vtrpcpb.Code_UNAVAILABLE {
				cancel()
				return err
			}

			// There was no error. We have to see if we need to retry.
			// If context was canceled, likely due to client disconnect,
			// return normally without retrying.
			select {
			case <-ctx.Done():
				return nil
			default:
			}
			firstErrorTimeStamp := lastErrors.Record(rs.Target)
			if time.Since(firstErrorTimeStamp) >= *messageStreamGracePeriod {
				// Cancel all streams and return an error.
				cancel()
				return vterrors.Errorf(vtrpcpb.Code_DEADLINE_EXCEEDED, "message stream from %v has repeatedly failed for longer than %v", rs.Target, *messageStreamGracePeriod)
			}

			// It's not been too long since our last good send. Wait and retry.
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(*messageStreamGracePeriod / 5):
			}
		}
	})
	return allErrors.AggrError(vterrors.Aggregate)
}
499318
func (stream *Stream) connect() {

	if stream.streamConnection.retries == stream.streamConnection.currentRetries {
		stream.Close()
		stream.gitter.log("Number of retries exceeded the max retries number, we are done here")
		return
	}

	res, err := stream.gitter.getResponse(stream.url, stream)
	if err != nil || res.StatusCode != 200 {
		stream.gitter.log("Failed to get response, trying reconnect")
		if res != nil {
			stream.gitter.log(fmt.Sprintf("Status code: %v", res.StatusCode))
		}
		stream.gitter.log(err)

		// sleep and wait
		stream.streamConnection.currentRetries++
		time.Sleep(time.Millisecond * stream.streamConnection.wait * time.Duration(stream.streamConnection.currentRetries))

		// connect again
		stream.Close()
		stream.connect()
	} else {
		stream.gitter.log("Response was received")
		stream.streamConnection.currentRetries = 0
		stream.streamConnection.closed = false
		stream.streamConnection.response = res
	}
}
539858
func (c *channelPool) getConns() chan *idleConn {
	c.mu.Lock()
	conns := c.conns
	c.mu.Unlock()
	return conns
}
1320543
func (h *httpStreamHandler) run() {
	klog.V(5).Infof("(conn=%p) waiting for port forward streams", h.conn)
Loop:
	for {
		select {
		case <-h.conn.CloseChan():
			klog.V(5).Infof("(conn=%p) upgraded connection closed", h.conn)
			break Loop
		case stream := <-h.streamChan:
			requestID := h.requestID(stream)
			streamType := stream.Headers().Get(api.StreamType)
			klog.V(5).Infof("(conn=%p, request=%s) received new stream of type %s", h.conn, requestID, streamType)

			p, created := h.getStreamPair(requestID)
			if created {
				go h.monitorStreamPair(p, time.After(h.streamCreationTimeout))
			}
			if complete, err := p.add(stream); err != nil {
				msg := fmt.Sprintf("error processing stream for request %s: %v", requestID, err)
				utilruntime.HandleError(errors.New(msg))
				p.printError(msg)
			} else if complete {
				go h.portForward(p)
			}
		}
	}
}
1242263
func (obj *Doc) GetConnections(ctx context.Context) ([]*Connection, error) {
	result := &struct {
		Connections []*Connection	`json:"qConnections"`
	} {}
	err := obj.rpc(ctx, "GetConnections", result)
	return result.Connections, err 
}
546288
func (r *rpcHandler) streamingRpc(server *serverParts, method string) (net.Conn, error) {
	// Try to dial the server
	conn, err := net.DialTimeout("tcp", server.Addr.String(), 10*time.Second)
	if err != nil {
		return nil, err
	}

	// Cast to TCPConn
	if tcp, ok := conn.(*net.TCPConn); ok {
		tcp.SetKeepAlive(true)
		tcp.SetNoDelay(true)
	}

	if err := r.streamingRpcImpl(conn, server.Region, method); err != nil {
		return nil, err
	}

	return conn, nil
}
378717
func (s *Swarm) StreamHandler() StreamHandler {
	s.handlerLock.RLock()
	defer s.handlerLock.RUnlock()
	if s.streamHandler == nil {
		return NoOpStreamHandler
	}
	return s.streamHandler
}