// Fork stream stuff here
356950
func (t *task) stream() {
	var consecutiveFailures int
	resetTime := time.Second * 3
	for {
		metricsChan, errChan, err := t.metricsManager.StreamMetrics(
			t.id,
			t.workflow.tags,
			t.maxCollectDuration,
			t.maxMetricsBuffer)
		if err != nil {
			consecutiveFailures++
			// check task failures
			if t.stopOnFailure >= 0 && consecutiveFailures >= t.stopOnFailure {
				taskLogger.WithFields(log.Fields{
					"_block":               "stream",
					"task-id":              t.id,
					"task-name":            t.name,
					"consecutive failures": consecutiveFailures,
					"error":                t.lastFailureMessage,
				}).Error(ErrTaskDisabledOnFailures)
				// disable the task
				t.disable(t.lastFailureMessage)
				return
			}
			// If we are unsuccessful at setting up the stream
			// wait for a second and then try again until either
			// the connection is successful or we pass the
			// acceptable number of consecutive failures
			time.Sleep(resetTime)
			continue
		} else {
			consecutiveFailures = 0
		}
		done := false
		for !done {
			if errChan == nil {
				break
			}
			select {
			case <-t.killChan:
				t.Lock()
				t.state = core.TaskStopped
				t.Unlock()
				done = true
				event := new(scheduler_event.TaskStoppedEvent)
				event.TaskID = t.id
				defer t.eventEmitter.Emit(event)
				return
			case mts, ok := <-metricsChan:
				if !ok {
					metricsChan = nil
					break
				}
				if len(mts) == 0 {
					continue
				}
				t.hitCount++
				consecutiveFailures = 0
				t.workflow.StreamStart(t, mts)
			case err := <-errChan:
				taskLogger.WithFields(log.Fields{
					"_block":    "stream",
					"task-id":   t.id,
					"task-name": t.name,
				}).Error("Error: " + err.Error())
				consecutiveFailures++
				if err.Error() == "connection broken" {
					// Wait here before trying to reconnect to allow time
					// for plugin restarts.
					time.Sleep(resetTime)
					done = true
				}
				// check task failures
				if t.stopOnFailure >= 0 && consecutiveFailures >= t.stopOnFailure {
					taskLogger.WithFields(log.Fields{
						"_block":               "stream",
						"task-id":              t.id,
						"task-name":            t.name,
						"consecutive failures": consecutiveFailures,
						"error":                t.lastFailureMessage,
					}).Error(ErrTaskDisabledOnFailures)
					// disable the task
					t.disable(t.lastFailureMessage)
					return
				}
			}
		}
	}
}
341583
func (c *Client) Stream(method, path string, in, out interface{}) (stream.Stream, error) {
	return c.StreamWithHeader(method, path, make(http.Header), in, out)
}
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
}
580388
func (p *Pan) Stream(samples [][2]float64) (n int, ok bool) {
	n, ok = p.Streamer.Stream(samples)
	switch {
	case p.Pan < 0:
		for i := range samples[:n] {
			r := samples[i][1]
			samples[i][0] += -p.Pan * r
			samples[i][1] -= -p.Pan * r
		}
	case p.Pan > 0:
		for i := range samples[:n] {
			l := samples[i][0]
			samples[i][0] -= p.Pan * l
			samples[i][1] += p.Pan * l
		}
	}
	return n, ok
}
54199
func streamEventCallback(st C.virStreamPtr, events int, callbackID int) {
	callbackFunc := getCallbackId(callbackID)

	callback, ok := callbackFunc.(StreamEventCallback)
	if !ok {
		panic("Incorrect stream event func callback")
	}

	callback(&Stream{ptr: st}, StreamEventType(events))
}
1302029
func (h *Handler) pipeUpstream(w http.ResponseWriter, r *cacheRequest) {
	rw := newResponseStreamer(w)
	rdr, err := rw.Stream.NextReader()
	if err != nil {
		debugf("error creating next stream reader: %v", err)
		w.Header().Set(CacheHeader, "SKIP")
		h.upstream.ServeHTTP(w, r.Request)
		return
	}
	defer rdr.Close()

	debugf("piping request upstream")
	go func() {
		h.upstream.ServeHTTP(rw, r.Request)
		rw.Stream.Close()
	}()
	rw.WaitHeaders()

	if r.Method != "HEAD" && !r.isStateChanging() {
		return
	}

	res := rw.Resource()
	defer res.Close()

	if r.Method == "HEAD" {
		h.cache.Freshen(res, r.Key.ForMethod("GET").String())
	} else if res.IsNonErrorStatus() {
		h.invalidateResource(res, r)
	}
}
112407
func main() {

	stream := health.NewStream().AddSink(&health.WriterSink{os.Stdout})
	stream.Event("wat")
	cleanKeyspace()

	queues := []string{"myqueue", "myqueue2", "myqueue3", "myqueue4", "myqueue5"}
	numJobs := 100000 / len(queues)

	job := stream.NewJob("enqueue_all")
	for _, q := range queues {
		enqueueJobs(q, numJobs)
	}
	job.Complete(health.Success)

	goworker.Register("MyClass", myJob)

	go monitor()

	// Blocks until process is told to exit via unix signal
	goworker.Work()
}
112654
func (c *conn) OpenStream() (smux.Stream, error) {
	s, err := c.spdyConn().CreateStream(http.Header{
		":method": []string{"POST"}, // this is here for HTTP/SPDY interop
		":path":   []string{"/"},    // this is here for HTTP/SPDY interop
	}, nil, false)
	if err != nil {
		return nil, err
	}

	// wait for a response before writing. for some reason
	// spdystream does not make forward progress unless you do this.
	s.Wait()
	return (*stream)(s), nil
}
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),
	}
}
1124200
func (s *Swarm) setupStream(smuxStream smux.Stream, c *Conn) *Stream {

	// create a new stream
	stream := newStream(smuxStream, c)

	// add it to our streams maps
	s.streamLock.Lock()
	c.streamLock.Lock()
	s.streams[stream] = struct{}{}
	c.streams[stream] = struct{}{}
	s.streamLock.Unlock()
	c.streamLock.Unlock()

	s.notifyAll(func(n Notifiee) {
		n.OpenedStream(stream)
	})
	return stream
}
433237
func (c *Client) DoStream(ctx context.Context, req *http.Request) (*http.Response, error) {
	resp, err := c.do(ctx, req)
	if err != nil {
		return nil, err
	}

	if resp.StatusCode != http.StatusOK {
		resp.Body.Close()
		return nil, getErrorFromResponse(resp)
	}

	return resp, nil
}
1110584
func forkAndMountat(dirfd uintptr, source, target, fstype, flags, data uintptr) (pid uintptr, errno syscall.Errno) {
	// block signal during clone
	beforeFork()

	// the cloned thread shares the open file descriptor, but the thread
	// never be reused by runtime.
	pid, _, errno = syscall.RawSyscall6(syscall.SYS_CLONE, uintptr(syscall.SIGCHLD)|syscall.CLONE_FILES, 0, 0, 0, 0, 0)
	if errno != 0 || pid != 0 {
		// restore all signals
		afterFork()
		return
	}

	// restore all signals
	afterForkInChild()

	// change working dir
	_, _, errno = syscall.RawSyscall(syscall.SYS_FCHDIR, dirfd, 0, 0)
	if errno != 0 {
		goto childerr
	}
	_, _, errno = syscall.RawSyscall6(syscall.SYS_MOUNT, source, target, fstype, flags, data, 0)

childerr:
	syscall.RawSyscall(syscall.SYS_EXIT, uintptr(errno), 0, 0)
	panic("unreachable")
}
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
}
858933
func flush(stream FSEventStreamRef, sync bool) {
	if sync {
		C.FSEventStreamFlushSync(stream)
	} else {
		C.FSEventStreamFlushAsync(stream)
	}
}
48192
func (s *stream) Header() {
	outputMutex.Lock()
	defer outputMutex.Unlock()
	s.log.header(s)
}
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)
}
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
}