// Stream starts streaming binlog events using the settings from NewStreamer().
438466
func (bls *Streamer) Stream(ctx context.Context) (err error) {
	// Ensure se is Open. If vttablet came up in a non_serving role,
	// the schema engine may not have been initialized.
	if err := bls.se.Open(); err != nil {
		return err
	}
	stopPos := bls.startPos
	defer func() {
		if err != nil && err != ErrBinlogUnavailable {
			err = fmt.Errorf("stream error @ %v: %v", stopPos, err)
		}
		log.Infof("stream ended @ %v, err = %v", stopPos, err)
	}()

	if bls.conn, err = NewSlaveConnection(bls.cp); err != nil {
		return err
	}
	defer bls.conn.Close()

	// Check that the default charsets match, if the client specified one.
	// Note that Streamer uses the settings for the 'dba' user, while
	// BinlogPlayer uses the 'filtered' user, so those are the ones whose charset
	// must match. Filtered replication should still succeed even with a default
	// mismatch, since we pass per-statement charset info. However, Vitess in
	// general doesn't support servers with different default charsets, so we
	// treat it as a configuration error.
	if bls.clientCharset != nil {
		cs, err := mysql.GetCharset(bls.conn.Conn)
		if err != nil {
			return fmt.Errorf("can't get charset to check binlog stream: %v", err)
		}
		log.Infof("binlog stream client charset = %v, server charset = %v", *bls.clientCharset, cs)
		if !proto.Equal(cs, bls.clientCharset) {
			return fmt.Errorf("binlog stream client charset (%v) doesn't match server (%v)", bls.clientCharset, cs)
		}
	}

	var events <-chan mysql.BinlogEvent
	if bls.timestamp != 0 {
		// MySQL 5.6 only: We are going to start reading the
		// logs from the beginning of a binlog file. That is
		// going to send us the PREVIOUS_GTIDS_EVENT that
		// contains the starting GTIDSet, and we will save
		// that as the current position.
		bls.usePreviousGTIDs = true
		events, err = bls.conn.StartBinlogDumpFromBinlogBeforeTimestamp(ctx, bls.timestamp)
	} else if !bls.startPos.IsZero() {
		// MySQL 5.6 only: we are starting from a random
		// binlog position. It turns out we will receive a
		// PREVIOUS_GTIDS_EVENT event, that has a GTIDSet
		// extracted from the binlogs. It is not related to
		// the starting position we pass in, it seems it is
		// just the PREVIOUS_GTIDS_EVENT from the file we're reading.
		// So we have to skip it.
		events, err = bls.conn.StartBinlogDumpFromPosition(ctx, bls.startPos)
	} else {
		bls.startPos, events, err = bls.conn.StartBinlogDumpFromCurrent(ctx)
	}
	if err != nil {
		return err
	}
	// parseEvents will loop until the events channel is closed, the
	// service enters the SHUTTING_DOWN state, or an error occurs.
	stopPos, err = bls.parseEvents(ctx, events)
	return err
}
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
}
1018793
func (l *bunyanLog) writeStream(stream Stream, data []byte) error {
	stream.Stream.Write(data)
	return nil
}
93910
func (c *Client) Stream(serviceMethod string, args interface{}) (io.ReadCloser, error) {
	var buf bytes.Buffer
	if err := json.NewEncoder(&buf).Encode(args); err != nil {
		return nil, err
	}
	return c.callWithRetry(serviceMethod, &buf, true)
}
694194
func (sf StreamerFunc) Stream(samples [][2]float64) (n int, ok bool) {
	return sf(samples)
}
393465
func newStream(client *http.Client, req *http.Request) *Stream {
	s := &Stream{
		client:   client,
		Messages: make(chan interface{}),
		done:     make(chan struct{}),
		group:    &sync.WaitGroup{},
	}
	s.group.Add(1)
	go s.retry(req, newExponentialBackOff(), newAggressiveExponentialBackOff())
	return s
}
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
	}
}
446616
func (es *EventStream) Stop() {
	if es.stream != nil {
		stop(es.stream, es.rlref)
		es.stream = nil
	}

	// Remove eventstream from the registry
	registry.Delete(es.registryID)
	es.registryID = 0
}
159253
func (h *BasicHost) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (inet.Stream, error) {
	pref, err := h.preferredProtocol(p, pids)
	if err != nil {
		return nil, err
	}

	if pref != "" {
		return h.newStream(ctx, p, pref)
	}

	var protoStrs []string
	for _, pid := range pids {
		protoStrs = append(protoStrs, string(pid))
	}

	s, err := h.Network().NewStream(ctx, p)
	if err != nil {
		return nil, err
	}

	selected, err := msmux.SelectOneOf(protoStrs, s)
	if err != nil {
		s.Reset()
		return nil, err
	}
	selpid := protocol.ID(selected)
	s.SetProtocol(selpid)
	h.Peerstore().AddProtocols(p, selected)

	return s, nil
}
100846
func (h *HTTPBin) Stream(w http.ResponseWriter, r *http.Request) {
	parts := strings.Split(r.URL.Path, "/")
	if len(parts) != 3 {
		http.Error(w, "Not found", http.StatusNotFound)
		return
	}
	n, err := strconv.Atoi(parts[2])
	if err != nil {
		http.Error(w, "Invalid integer", http.StatusBadRequest)
		return
	}

	if n > 100 {
		n = 100
	} else if n < 1 {
		n = 1
	}

	resp := &streamResponse{
		Args:    r.URL.Query(),
		Headers: getRequestHeaders(r),
		Origin:  getOrigin(r),
		URL:     getURL(r).String(),
	}

	f := w.(http.Flusher)
	for i := 0; i < n; i++ {
		resp.ID = i
		line, _ := json.Marshal(resp)
		w.Write(line)
		w.Write([]byte("\n"))
		f.Flush()
	}
}
341583
func (c *Client) Stream(method, path string, in, out interface{}) (stream.Stream, error) {
	return c.StreamWithHeader(method, path, make(http.Header), in, out)
}
1013547
func (m *Streamer) Stream(stdout, stderr chan []byte) StreamID {
	sid := m.idGen.next()
	m.streams.Store(sid, newStream(stdout, stderr))

	return sid
}
887655
func (e *streamExecutor) Stream(options StreamOptions) error {
	req, err := http.NewRequest(e.method, e.url.String(), nil)
	if err != nil {
		return fmt.Errorf("error creating request: %v", err)
	}

	conn, protocol, err := spdy.Negotiate(
		e.upgrader,
		&http.Client{Transport: e.transport},
		req,
		e.protocols...,
	)
	if err != nil {
		return err
	}
	defer conn.Close()

	var streamer streamProtocolHandler

	switch protocol {
	case remotecommand.StreamProtocolV4Name:
		streamer = newStreamProtocolV4(options)
	case remotecommand.StreamProtocolV3Name:
		streamer = newStreamProtocolV3(options)
	case remotecommand.StreamProtocolV2Name:
		streamer = newStreamProtocolV2(options)
	case "":
		klog.V(4).Infof("The server did not negotiate a streaming protocol version. Falling back to %s", remotecommand.StreamProtocolV1Name)
		fallthrough
	case remotecommand.StreamProtocolV1Name:
		streamer = newStreamProtocolV1(options)
	}

	return streamer.stream(conn)
}
895093
func (c *RPCClient) Stream(filter string, ch chan<- map[string]interface{}) (StreamHandle, error) {
	// Setup the request
	seq := c.getSeq()
	header := requestHeader{
		Command: streamCommand,
		Seq:     seq,
	}
	req := streamRequest{
		Type: filter,
	}

	// Create a monitor handler
	initCh := make(chan error, 1)
	handler := &streamHandler{
		client:  c,
		initCh:  initCh,
		eventCh: ch,
		seq:     seq,
	}
	c.handleSeq(seq, handler)

	// Send the request
	if err := c.send(&header, &req); err != nil {
		c.deregisterHandler(seq)
		return 0, err
	}

	// Wait for a response
	select {
	case err := <-initCh:
		return StreamHandle(seq), err
	case <-c.shutdownCh:
		c.deregisterHandler(seq)
		return 0, clientClosed
	}
}
1009490
func (b *Butler) Streams() []types.StreamName {
	var streams types.StreamNameSlice
	func() {
		b.streamSeenLock.Lock()
		defer b.streamSeenLock.Unlock()

		streams = make([]types.StreamName, 0, b.streamSeen.Len())
		b.streamSeen.Iter(func(s string) bool {
			streams = append(streams, types.StreamName(s))
			return true
		})
	}()

	sort.Sort(streams)
	return ([]types.StreamName)(streams)
}
805015
func (conn *gRPCQueryClient) VStream(ctx context.Context, target *querypb.Target, position string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error {
	stream, err := func() (queryservicepb.Query_VStreamClient, error) {
		conn.mu.RLock()
		defer conn.mu.RUnlock()
		if conn.cc == nil {
			return nil, tabletconn.ConnClosed
		}

		req := &binlogdatapb.VStreamRequest{
			Target:            target,
			EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
			ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
			Position:          position,
			Filter:            filter,
		}
		stream, err := conn.c.VStream(ctx, req)
		if err != nil {
			return nil, tabletconn.ErrorFromGRPC(err)
		}
		return stream, nil
	}()
	if err != nil {
		return err
	}
	for {
		r, err := stream.Recv()
		if err != nil {
			return tabletconn.ErrorFromGRPC(err)
		}
		select {
		case <-ctx.Done():
			return nil
		default:
		}
		if err := send(r.Events); err != nil {
			if err == io.EOF {
				return nil
			}
			return err
		}
	}
}
1018384
func (s *SelectObjectContentOutput) SetEventStream(v *SelectObjectContentEventStream) *SelectObjectContentOutput {
	s.EventStream = v
	return s
}