// Stream starts a new stream.
425289
func (vse *Engine) Stream(ctx context.Context, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error {
	// Ensure kschema is initialized and the watcher is started.
	// Starting of the watcher has to be delayed till the first call to Stream
	// because this overhead should be incurred only if someone uses this feature.
	vse.watcherOnce.Do(vse.setWatch)

	// Create stream and add it to the map.
	streamer, idx, err := func() (*vstreamer, int, error) {
		vse.mu.Lock()
		defer vse.mu.Unlock()
		if !vse.isOpen {
			return nil, 0, errors.New("VStreamer is not open")
		}
		streamer := newVStreamer(ctx, vse.cp, vse.se, startPos, filter, vse.kschema, send)
		idx := vse.streamIdx
		vse.streamers[idx] = streamer
		vse.streamIdx++
		// Now that we've added the stream, increment wg.
		// This must be done before releasing the lock.
		vse.wg.Add(1)
		return streamer, idx, nil
	}()
	if err != nil {
		return err
	}

	// Remove stream from map and decrement wg when it ends.
	defer func() {
		vse.mu.Lock()
		defer vse.mu.Unlock()
		delete(vse.streamers, idx)
		vse.wg.Done()
	}()

	// No lock is held while streaming, but wg is incremented.
	return streamer.Stream()
}
138081
func (s *stream) Start() error {
	if s.ref != nilstream {
		return nil
	}
	wg.Wait()
	p := C.CFStringCreateWithCStringNoCopy(C.kCFAllocatorDefault, C.CString(s.path), C.kCFStringEncodingUTF8, C.kCFAllocatorDefault)
	path := C.CFArrayCreate(C.kCFAllocatorDefault, (*unsafe.Pointer)(unsafe.Pointer(&p)), 1, nil)
	ctx := C.FSEventStreamContext{}
	ref := C.EventStreamCreate(&ctx, C.uintptr_t(s.info), path, C.FSEventStreamEventId(atomic.LoadUint64(&since)), latency, flags)
	if ref == nilstream {
		return errCreate
	}
	C.FSEventStreamScheduleWithRunLoop(ref, runloop, C.kCFRunLoopDefaultMode)
	if C.FSEventStreamStart(ref) == C.Boolean(0) {
		C.FSEventStreamInvalidate(ref)
		return errStart
	}
	C.CFRunLoopWakeUp(runloop)
	s.ref = ref
	return nil
}
215175
func (l *Logger) AddStream(stream Stream) error {
	if err := stream.init(l.config); err != nil {
		return err
	}
	l.streams = append(l.streams, stream)
	return nil
}
419914
func (s *Stream) Write() error {
	if s.callback.IsValid() {
		return CanNotWriteToACallbackStream
	}
	if s.out == nil {
		return CanNotWriteToAnInputOnlyStream
	}
	buf, frames, err := getBuffer(s.out, s.outParams)
	if err != nil {
		return err
	}
	return newError(C.Pa_WriteStream(s.paStream, buf, C.ulong(frames)))
}
323644
func NewStream(w io.Writer) *Stream {
	return &Stream{
		Writer: w,
		enc:    json.NewEncoder(w),
	}
}
1293667
func NewStream(reader io.Reader, writer io.Writer, maxWriteDelay time.Duration) *Stream {
	return &Stream{
		Decoder: NewDecoder(reader),
		Encoder: NewEncoder(writer, maxWriteDelay),
	}
}
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
	}
}
140935
func (s *Snapshotter) stream() {
	clockTicker := time.NewTicker(clockUpdateInterval)
	defer clockTicker.Stop()

	// flushEvent is used to handle writing out an event
	flushEvent := func(e Event) {
		// Stop recording events after a leave is issued
		if s.leaving {
			return
		}
		switch typed := e.(type) {
		case MemberEvent:
			s.processMemberEvent(typed)
		case UserEvent:
			s.processUserEvent(typed)
		case *Query:
			s.processQuery(typed)
		default:
			s.logger.Printf("[ERR] serf: Unknown event to snapshot: %#v", e)
		}
	}

	for {
		select {
		case <-s.leaveCh:
			s.leaving = true

			// If we plan to re-join, keep our state
			if !s.rejoinAfterLeave {
				s.aliveNodes = make(map[string]string)
			}
			s.tryAppend("leave\n")
			if err := s.buffered.Flush(); err != nil {
				s.logger.Printf("[ERR] serf: failed to flush leave to snapshot: %v", err)
			}
			if err := s.fh.Sync(); err != nil {
				s.logger.Printf("[ERR] serf: failed to sync leave to snapshot: %v", err)
			}

		case e := <-s.streamCh:
			flushEvent(e)

		case <-clockTicker.C:
			s.updateClock()

		case <-s.shutdownCh:
			// Setup a timeout
			flushTimeout := time.After(shutdownFlushTimeout)

			// Snapshot the clock
			s.updateClock()

			// Clear out the buffers
		FLUSH:
			for {
				select {
				case e := <-s.streamCh:
					flushEvent(e)
				case <-flushTimeout:
					break FLUSH
				default:
					break FLUSH
				}
			}

			if err := s.buffered.Flush(); err != nil {
				s.logger.Printf("[ERR] serf: failed to flush snapshot: %v", err)
			}
			if err := s.fh.Sync(); err != nil {
				s.logger.Printf("[ERR] serf: failed to sync snapshot: %v", err)
			}
			s.fh.Close()
			close(s.waitCh)
			return
		}
	}
}
790300
func (a *Adapter) Stream(logstream chan *router.Message) {
	for message := range logstream {
		buf := new(bytes.Buffer)
		err := a.tmpl.Execute(buf, message)
		if err != nil {
			log.Println("raw:", err)
			return
		}
		//log.Println("debug:", buf.String())
		_, err = a.conn.Write(buf.Bytes())
		if err != nil {
			log.Println("raw:", err)
			if reflect.TypeOf(a.conn).String() != "*net.UDPConn" {
				return
			}
		}
	}
}
162848
func (r *RestartableResultReader) startStream() (bool, error) {
	// Start the streaming query.
	r.generateQuery()
	var stream sqltypes.ResultStream

	if r.txID == 0 {
		stream = queryservice.ExecuteWithStreamer(r.ctx, r.conn, &querypb.Target{
			Keyspace:   r.tablet.Keyspace,
			Shard:      r.tablet.Shard,
			TabletType: r.tablet.Type,
		}, r.query, make(map[string]*querypb.BindVariable), nil)
	} else {
		stream = queryservice.ExecuteWithTransactionalStreamer(r.ctx, r.conn, &querypb.Target{
			Keyspace:   r.tablet.Keyspace,
			Shard:      r.tablet.Shard,
			TabletType: r.tablet.Type,
		}, r.query, make(map[string]*querypb.BindVariable), r.txID, nil)
	}

	// Read the fields information.
	cols, err := stream.Recv()
	if err != nil {
		return true /* retryable */, vterrors.Wrapf(err, "cannot read Fields for query '%v'", r.query)
	}
	r.fields = cols.Fields
	r.output = stream

	alias := topoproto.TabletAliasString(r.tablet.Alias)
	statsStreamingQueryCounters.Add(alias, 1)
	log.V(2).Infof("tablet=%v table=%v chunk=%v: Starting to stream rows using query '%v'.", alias, r.td.Name, r.chunk, r.query)
	return false, nil
}
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)
}
981407
func (v *Volume) Stream(samples [][2]float64) (n int, ok bool) {
	n, ok = v.Streamer.Stream(samples)
	gain := 0.0
	if !v.Silent {
		gain = math.Pow(v.Base, v.Volume)
	}
	for i := range samples[:n] {
		samples[i][0] *= gain
		samples[i][1] *= gain
	}
	return n, ok
}
346371
func (f *firebaseAPI) Stream(path, auth string, body interface{}, params map[string]string, stop <-chan bool) (<-chan RawEvent, error) {
	response, err := doFirebaseRequest(streamClient, "GET", path, auth,
		"text/event-stream", body, params)
	if err != nil {
		return nil, err
	}

	go func() {
		<-stop
		response.Body.Close()
	}()

	events := make(chan RawEvent, 1000)

	// bufio.Scanner barfs on really long events, as its buffer size is pretty
	// small, and it can't be overridden. This is not the most memory-optimal
	// implementation of the streaming client, but each event is not expected
	// to exceed several KB.
	go func() {
		var (
			err       error
			firstLine string
			lineBuf   []byte
		)

		byteReader := bufio.NewReader(response.Body)

		for {
			var b byte

			b, err = byteReader.ReadByte()
			if err != nil {
				break
			}

			if b != "\n"[0] {
				lineBuf = append(lineBuf, b)
				continue
			}

			if firstLine == "" {
				firstLine = string(lineBuf)
				lineBuf = []byte{}
				continue
			}

			line := string(lineBuf)

			event := RawEvent{}
			event.Event = strings.Replace(firstLine, "event: ", "", 1)
			event.Data = strings.Replace(line, "data: ", "", 1)

			events <- event
			firstLine = ""
			lineBuf = []byte{}
		}

		if err == io.EOF {
			err = nil
		}

		closeEvent := RawEvent{Error: err}
		events <- closeEvent
		close(events)
	}()

	return events, nil
}
1028479
func (c *Client) startStreaming(ctx context.Context, path string) (*http.Response, error) {
	req, err := c.NewStreamingRequest(path, nil)
	if err != nil {
		return nil, err
	}

	resp, err := c.DoStream(ctx, req)
	if err != nil {
		return nil, err
	}

	return resp, nil
}
999424
func (c *client) Stream(id int64, rc io.ReadCloser) error {
	uri := fmt.Sprintf(pathStream, c.base, id)
	err := c.post(uri, rc, nil)

	return err
}
380086
func (m *Streamer) Stop(streamID StreamID) {
	strm, ok := m.loadStream(streamID)
	if !ok {
		return
	}

	close(strm.done)

	go func() {
		// wait some time to ensure clients have connected, once they've
		// retrieved the stream from the map it's safe to delete the key
		time.Sleep(m.graceTime)

		m.streams.Delete(streamID)
	}()
}
1214295
func stream(client *http.Client, rawurl, method string, in, out interface{}) (io.ReadCloser, error) {
	uri, err := url.Parse(rawurl)
	if err != nil {
		return nil, err
	}

	// if we are posting or putting data, we need to
	// write it to the body of the request.
	var buf io.ReadWriter
	if in == nil {
		// nothing
	} else if rw, ok := in.(io.ReadWriter); ok {
		buf = rw
	} else if b, ok := in.([]byte); ok {
		buf = new(bytes.Buffer)
		buf.Write(b)
	} else {
		buf = new(bytes.Buffer)
		err := json.NewEncoder(buf).Encode(in)
		if err != nil {
			return nil, err
		}
	}

	// creates a new http request to bitbucket.
	req, err := http.NewRequest(method, uri.String(), buf)
	if err != nil {
		return nil, err
	}
	if in == nil {
		// nothing
	} else if _, ok := in.(io.ReadWriter); ok {
		req.Header.Set("Content-Type", "plain/text")
	} else {
		req.Header.Set("Content-Type", "application/json")
	}

	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode > http.StatusPartialContent {
		defer resp.Body.Close()
		out, _ := ioutil.ReadAll(resp.Body)
		return nil, fmt.Errorf(string(out))
	}
	return resp.Body, nil
}