// Stream returns a stream.Stream for a specific method and path. in is an
// optional json object to be sent to the server via the body, and out is a
// required channel, to which the output will be streamed.
341583
func (c *Client) Stream(method, path string, in, out interface{}) (stream.Stream, error) {
	return c.StreamWithHeader(method, path, make(http.Header), in, out)
}
205612
func (stream *Stream) Write(p []byte) (nn int, err error) {
	stream.buf = append(stream.buf, p...)
	if stream.out != nil {
		nn, err = stream.out.Write(stream.buf)
		stream.buf = stream.buf[nn:]
		return
	}
	return len(p), nil
}
1277316
func InStream(pattern string, stream io.Reader) *list.List {
	var (
		scanner = bufio.NewScanner(stream)
		sp      = NewShellPattern("*" + pattern + "*")
		line    string
		lineNo  int
		result  = list.New()
	)

	for scanner.Scan() {
		line = scanner.Text()
		if sp.Match(line) {
			result.PushBack(&Ref{lineNo, line})
		}
		lineNo++
	}
	return result
}
147678
func (s StreamReadError) Error() string {
	return fmt.Sprintf("error reading stream %d: %s", s.Stream, s.Err)
}
346839
func newStream(path string, fn streamFunc) *stream {
	return &stream{
		path: path,
		info: streamFuncs.add(fn),
	}
}
273827
func (c *Client) processStream(baseURL *url.URL, path string, authHandler AuthHandler, callb func(interface{}) error, data interface{}) error {
	var requestUrl string

	if q := strings.Index(path, "?"); q > 0 {
		requestUrl = baseURL.Scheme + "://" + baseURL.Host + path[:q] + "?" + path[q+1:]
	} else {
		requestUrl = baseURL.Scheme + "://" + baseURL.Host + path
	}

	req, err := http.NewRequest("GET", requestUrl, nil)
	if err != nil {
		return err
	}

	err = maybeAddAuth(req, authHandler)
	if err != nil {
		return err
	}

	res, err := doHTTPRequest(req)
	if err != nil {
		return err
	}

	defer res.Body.Close()
	if res.StatusCode != 200 {
		bod, _ := ioutil.ReadAll(io.LimitReader(res.Body, 512))
		return fmt.Errorf("HTTP error %v getting %q: %s",
			res.Status, requestUrl, bod)
	}

	reader := bufio.NewReader(res.Body)
	for {
		bs, err := reader.ReadBytes('\n')
		if err != nil {
			return err
		}
		if len(bs) == 1 && bs[0] == '\n' {
			continue
		}

		err = json.Unmarshal(bs, data)
		if err != nil {
			return err
		}
		err = callb(data)
		if err != nil {
			return err
		}
	}
	return nil

}
886756
func (c *Ctrl) Stream(samples [][2]float64) (n int, ok bool) {
	if c.Streamer == nil {
		return 0, false
	}
	if c.Paused {
		for i := range samples {
			samples[i] = [2]float64{}
		}
		return len(samples), true
	}
	return c.Streamer.Stream(samples)
}
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
		}
	}
}
1040214
func Stream(stream bool) optSetter {
	return func(f *Forwarder) error {
		f.stream = stream
		return nil
	}
}
638526
func (rng *Pcg128x64) Stream(streamH, streamL uint64) {
  // stream must be odd for LCG, so shift left 1 and turn on the 1 bit
  rng.lcg.Stream = Uint128{streamH,streamL}.ShiftLeft(1)
  rng.lcg.Stream.L |= 1
}
1298533
func (s *Stream) Drain() (error) {
	var err C.int
	_ = C.pa_simple_drain(s.simple, &err)
	if err == C.PA_OK {
		return nil
	}
	return errorFromCode(err)
}
218465
func (s *Stream) String() (str string) {
	return fmt.Sprintf(
		"RtmpPublishUrl:%s\n RtmpLiveUrls:%s\n  HlsLiveUrls:%s\n  HttpFlvLiveUrls:%s\n  OnlineUserNum:%d \n",
		s.rtmpPublishUrl,
		s.rtmpLiveUrls,
		s.hlsLiveUrls,
		s.httpFlvLiveUrls,
		s.OnlineUserNum(),
	)
}
1241785
func (p StreamPath) MarshalJSON() ([]byte, error) {
	v := string(p)
	return json.Marshal(&v)
}
896197
func (a *Adapter) Stream(logstream chan *router.Message) {
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		a.subAdapter.Stream(a.out)
		wg.Done()
	}()
	defer func() {
		for _, message := range a.buffers {
			a.out <- message
		}

		close(a.out)
		wg.Wait()
	}()

	for {
		select {
		case message, ok := <-logstream:
			if !ok {
				return
			}

			if !multilineContainer(message.Container, a.enableByDefault) {
				a.out <- message
				continue
			}

			cID := message.Container.ID
			old, oldExists := a.buffers[cID]
			if a.isFirstLine(message) {
				if oldExists {
					a.out <- old
				}

				a.buffers[cID] = message
			} else {
				isLastLine := a.isLastLine(message)
				
				if oldExists {
					old.Data += a.separator + message.Data
					message = old
				}

				if isLastLine {
					a.out <- message
					if oldExists {
						delete(a.buffers, cID)
					}
				} else {
					a.buffers[cID] = message
				}
			}
		case <-a.nextCheck:
			now := time.Now()

			for key, message := range a.buffers {
				if message.Time.Add(a.flushAfter).After(now) {
					a.out <- message
					delete(a.buffers, key)
				}
			}

			a.nextCheck = time.After(a.checkInterval)
		}
	}
}
1347966
func (c *Context) Stream(step func(w io.Writer) bool) bool {
	w := c.Writer
	clientGone := w.CloseNotify()
	for {
		select {
		case <-clientGone:
			return true
		default:
			keepOpen := step(w)
			w.Flush()
			if !keepOpen {
				return false
			}
		}
	}
}
674854
func (c *Ctx) Stream(step func(w io.Writer) bool) {
	w := c.response
	clientGone := w.CloseNotify()

	for {
		select {
		case <-clientGone:
			return
		default:
			keepOpen := step(w)
			w.Flush()
			if !keepOpen {
				return
			}
		}
	}
}
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)))
}