// Stream.Latency returns the playback latency in microseconds.
188892
func (s *Stream) Latency() (uint64, error) {
	var err C.int
	lat := C.pa_simple_get_latency(s.simple, &err)
	if err == C.PA_OK {
		return uint64(lat), nil
	}
	return uint64(lat), errorFromCode(err)
}
667651
func (vs *vstreamer) Stream() error {
	defer vs.cancel()

	pos, err := mysql.DecodePosition(vs.startPos)
	if err != nil {
		return err
	}
	vs.pos = pos

	// Ensure se is Open. If vttablet came up in a non_serving role,
	// the schema engine may not have been initialized.
	if err := vs.se.Open(); err != nil {
		return wrapError(err, vs.pos)
	}

	conn, err := binlog.NewSlaveConnection(vs.cp)
	if err != nil {
		return wrapError(err, vs.pos)
	}
	defer conn.Close()

	events, err := conn.StartBinlogDumpFromPosition(vs.ctx, vs.pos)
	if err != nil {
		return wrapError(err, vs.pos)
	}
	err = vs.parseEvents(vs.ctx, events)
	return wrapError(err, vs.pos)
}
646355
func (v *Stats) WriteLatency() uint64 {
	if v.Writes == 0 {
		return 0
	}
	return (uint64)((v.WriteMs * 1000) / v.Writes)
}
361235
func HighLatencyParameters(in, out *DeviceInfo) (p StreamParameters) {
	sampleRate := 0.0
	if in != nil {
		p := &p.Input
		p.Device = in
		p.Channels = 1
		if in.MaxInputChannels < 1 {
			p.Channels = in.MaxInputChannels
		}
		p.Latency = in.DefaultHighInputLatency
		sampleRate = in.DefaultSampleRate
	}
	if out != nil {
		p := &p.Output
		p.Device = out
		p.Channels = 2
		if out.MaxOutputChannels < 2 {
			p.Channels = out.MaxOutputChannels
		}
		p.Latency = out.DefaultHighOutputLatency
		if r := out.DefaultSampleRate; r < sampleRate || sampleRate == 0 {
			sampleRate = r
		}
	}
	p.SampleRate = sampleRate
	p.FramesPerBuffer = FramesPerBufferUnspecified
	return p
}
614854
func (s *PlayerLatency) SetLatencyInMilliseconds(v float64) *PlayerLatency {
	s.LatencyInMilliseconds = &v
	return s
}
703890
func (w *audioWriter) streamWriter(stream *portAudioStream) {
	defer w.wg.Done()
	defer stream.Close()

	buffer := make([]int16, audioOutputBufferSize)
	output := buffer[:]

	for {
		// Wait for input data or signal to quit.
		var input audio
		select {
		case input = <-w.input:
		case <-w.quit:
			return
		}

		// Initialize the audio stream based on the specification of the input format.
		err := stream.Stream(&output, input.format.Channels, input.format.SampleRate)
		if err != nil {
			panic(err)
		}

		// Decode the incoming data which is expected to be 2 channels and
		// delivered as int16 in []byte, hence we need to convert it.
		i := 0
		for i < len(input.frames) {
			j := 0
			for j < len(buffer) && i < len(input.frames) {
				buffer[j] = int16(input.frames[i]) | int16(input.frames[i+1])<<8
				j += 1
				i += 2
			}

			output = buffer[:j]
			stream.Write()
		}
	}
}
527175
func HostStreamGetInfo(context *gin.Context) {
	rateLimit, convertErr := util.ConvertStringToTimeDuration(context.Param("rateLimit"))
	context.Stream(func(w io.Writer) bool {
		if convertErr != nil {
			return false
		}
		time.Sleep(rateLimit * time.Second)
		body, err := host.ServiceGetInfo()
		return util.StreamHandleResponse(context, body, err, "HostStreamGetInfo")
	})
}
960454
func (h *HDRHistogram) LatencyAtQuantile(q float64) time.Duration {
	return time.Duration(h.ValueAtQuantile(q)) * time.Microsecond
}
1299108
func (l *Log) Stream(header string) Stream {
	return &stream{
		header: header,
		log:    l,
		quiet:  l.quiet,
	}
}
790476
func (c *StatsdClient) Timing(stat string, delta int64) error {
	return c.TimingWithSampling(stat, delta, 1)
}
630299
func (e *streamExecutor) Stream(stdin io.Reader, stdout, stderr io.Writer, tty bool) error {
	supportedProtocols := []string{StreamProtocolV2Name, StreamProtocolV1Name}
	conn, protocol, err := e.Dial(supportedProtocols...)
	if err != nil {
		return err
	}
	defer conn.Close()

	var streamer streamProtocolHandler

	switch protocol {
	case StreamProtocolV2Name:
		streamer = &streamProtocolV2{
			stdin:  stdin,
			stdout: stdout,
			stderr: stderr,
			tty:    tty,
		}
	case "":
		glog.V(4).Infof("The server did not negotiate a streaming protocol version. Falling back to %s", StreamProtocolV1Name)
		fallthrough
	case StreamProtocolV1Name:
		streamer = &streamProtocolV1{
			stdin:  stdin,
			stdout: stdout,
			stderr: stderr,
			tty:    tty,
		}
	}

	return streamer.stream(conn)
}
1220733
func (s *Session) AcceptStream() (*Stream, error) {
	var deadline <-chan time.Time
	if d, ok := s.deadline.Load().(time.Time); ok && !d.IsZero() {
		timer := time.NewTimer(time.Until(d))
		defer timer.Stop()
		deadline = timer.C
	}
	select {
	case stream := <-s.chAccepts:
		return stream, nil
	case <-deadline:
		return nil, errTimeout
	case <-s.die:
		return nil, errors.New(errBrokenPipe)
	}
}
341583
func (c *Client) Stream(method, path string, in, out interface{}) (stream.Stream, error) {
	return c.StreamWithHeader(method, path, make(http.Header), in, out)
}
776137
func (s *NoopClient) Timing(stat string, delta int64, rate float32) error {
	return nil
}
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)
		}
	}
}
758097
func (gitter *Gitter) Stream(roomID string) *Stream {
	return &Stream{
		url:    streamBaseURL + "rooms/" + roomID + "/chatMessages",
		Event:  make(chan Event),
		gitter: gitter,
		streamConnection: gitter.newStreamConnection(
			defaultConnectionWaitTime,
			defaultConnectionMaxRetries),
	}
}
281706
func (p *Player) Current() time.Duration {
	sample := int64(0)
	p.sync(func() {
		sample = p.pos / bytesPerSample / channelNum
	})
	return time.Duration(sample) * time.Second / time.Duration(p.sampleRate)
}