// MessageStream streams messages from a message table.
767687
func (qre *QueryExecutor) MessageStream(callback func(*sqltypes.Result) error) error {
	qre.logStats.OriginalSQL = qre.query
	qre.logStats.PlanType = qre.plan.PlanID.String()

	defer func(start time.Time) {
		tabletenv.QueryStats.Record(qre.plan.PlanID.String(), start)
		tabletenv.RecordUserQuery(qre.ctx, qre.plan.TableName(), "MessageStream", int64(time.Since(start)))
	}(time.Now())

	if err := qre.checkPermissions(); err != nil {
		return err
	}

	done, err := qre.tsv.messager.Subscribe(qre.ctx, qre.plan.TableName().String(), func(r *sqltypes.Result) error {
		select {
		case <-qre.ctx.Done():
			return io.EOF
		default:
		}
		return callback(r)
	})
	if err != nil {
		return err
	}
	<-done
	return nil
}
1169326
func NewStream(ctx context.Context, method string, conn *Conn) (Stream, error) {
	if ctx == nil {
		ctx = context.Background()
	}

	return newStreamClient(ctx, method, conn)
}
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
}
493796
func (tag PluginStructTag) GetStream() MessageStreamID {
	tagValue, tagSet := reflect.StructTag(tag).Lookup(PluginStructTagDefault)
	if !tagSet {
		return InvalidStreamID
	}
	return GetStreamID(tagValue)
}
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)
}
769210
func (c *client) LogStream(id int64) (StreamWriter, error) {
	rawurl := fmt.Sprintf(pathLogsAuth, c.base, id, c.token)
	uri, err := url.Parse(rawurl)
	if err != nil {
		return nil, err
	}
	if uri.Scheme == "https" {
		uri.Scheme = "wss"
	} else {
		uri.Scheme = "ws"
	}

	// TODO need TLS client configuration

	conn, _, err := websocket.DefaultDialer.Dial(uri.String(), nil)
	return conn, err
}
299565
func (tsv *TabletServer) MessageStream(ctx context.Context, target *querypb.Target, name string, callback func(*sqltypes.Result) error) (err error) {
	return tsv.execRequest(
		ctx, 0,
		"MessageStream", "stream", nil,
		target, nil, false /* isBegin */, false, /* allowOnShutdown */
		func(ctx context.Context, logStats *tabletenv.LogStats) error {
			plan, err := tsv.qe.GetMessageStreamPlan(name)
			if err != nil {
				return err
			}
			qre := &QueryExecutor{
				query:    "stream from msg",
				plan:     plan,
				ctx:      ctx,
				logStats: logStats,
				tsv:      tsv,
			}
			return qre.MessageStream(callback)
		},
	)
}
1211161
func (c *Cluster) GetWatchStream() chan *swarmapi.WatchMessage {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.watchStream
}
119149
func (dbc *DBConn) Stream(ctx context.Context, query string, callback func(*sqltypes.Result) error, streamBufferSize int, includedFields querypb.ExecuteOptions_IncludedFields) error {
	span, ctx := trace.NewSpan(ctx, "DBConn.Stream")
	trace.AnnotateSQL(span, query)
	defer span.Finish()

	resultSent := false
	for attempt := 1; attempt <= 2; attempt++ {
		err := dbc.streamOnce(
			ctx,
			query,
			func(r *sqltypes.Result) error {
				if !resultSent {
					resultSent = true
					r = r.StripMetadata(includedFields)
				}
				return callback(r)
			},
			streamBufferSize,
		)
		switch {
		case err == nil:
			// Success.
			return nil
		case !mysql.IsConnErr(err):
			// Not a connection error. Don't retry.
			return err
		case attempt == 2:
			// Reached the retry limit.
			return err
		case resultSent:
			// Don't retry if streaming has started.
			return err
		}

		// Connection error. Retry if context has not expired.
		select {
		case <-ctx.Done():
			return err
		default:
		}
		if reconnectErr := dbc.reconnect(); reconnectErr != nil {
			dbc.pool.checker.CheckMySQL()
			// Return the error of the reconnect and not the original connection error.
			return reconnectErr
		}
	}
	panic("unreachable")
}
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
}
323644
func NewStream(w io.Writer) *Stream {
	return &Stream{
		Writer: w,
		enc:    json.NewEncoder(w),
	}
}
1046636
func (g *GoWebSocket) MessageSend(v interface{}) error {
	return websocket.Message.Send(g.Conn, v)
}
596558
func (conn *VTGateConn) UpdateStream(ctx context.Context, keyspace, shard string, keyRange *topodatapb.KeyRange, tabletType topodatapb.TabletType, timestamp int64, event *querypb.EventToken) (UpdateStreamReader, error) {
	return conn.impl.UpdateStream(ctx, keyspace, shard, keyRange, tabletType, timestamp, event)
}
915623
func (reader *PluginConfigReader) GetStreamMap(key string, defaultValue string) map[MessageStreamID]string {
	value, err := reader.WithError.GetStreamMap(key, defaultValue)
	reader.Errors.Push(err)
	return value
}
433148
func NewMessageStream(conn net.Conn, parser Parser) *MessageStream {
	m := &MessageStream{
		conn,
		NewBufferPool(),
		parser,
		make(chan bool, 1), // parserShutdown
		0,
		make(chan error, 1),   // Error
		make(chan Message, 1), // Inbound
		make(chan Message, 1), // Outbound
		make(chan bool, 1),    // Shutdown
	}

	go m.outbound()
	go m.inbound()
	for i := 0; i < numParserGoroutines; i++ {
		go m.parse()
	}

	return m
}
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
}
779341
func (m *MessageStream) parse() {
	errMessage := "received: %v and encountered error: %v"
	for {
		select {
		case b := <-m.pool.Full:
			msg, err := m.parser.Parse(b.Bytes())
			// Log all message parsing errors.
			if err != nil {
				log.Errorf(errMessage, b.Bytes(), err)
			}

			m.Inbound <- msg
			b.Reset()
			m.pool.Empty <- b
		case <-m.parserShutdown:
			return
		}
	}
}