// waitStreamReply waits until either replySent or stop is closed. If replySent is closed, it sends
// an empty struct to the notify channel.
184438
func waitStreamReply(ctx context.Context, replySent <-chan struct{}, notify chan<- struct{}) {
	select {
	case <-replySent:
		notify <- struct{}{}
	case <-ctx.Done():
	}
}
1071412
func StreamHandleResponse(context *gin.Context, body interface{}, err error, messageName string) bool {
	if err != nil {
		return false
	}
	context.SSEvent(messageName, body)
	return true
}
146847
func waitServer(url string) bool {
	tries := 20
	for tries > 0 {
		resp, err := http.Get(url)
		if err == nil {
			resp.Body.Close()
			return true
		}
		time.Sleep(100 * time.Millisecond)
		tries--
	}
	return false
}
931242
func (c *AppStream) WaitUntilFleetStarted(input *DescribeFleetsInput) error {
	return c.WaitUntilFleetStartedWithContext(aws.BackgroundContext(), input)
}
213288
func ZmqReply(_socket *zmq.Socket, _compute_reply RecieveArrayReturnString) error {
	_msg, err := _socket.Recv(0)
	if err != nil {
		return err
	}

	_msg_array := strings.Fields(string(_msg))
	return_value := _compute_reply(_msg_array)

	err = _socket.Send([]byte(return_value), 0)
	return err
}
645665
func BroadcastWaitForResponse(responses chan common.Status, errs chan error) (common.Status, error) {
	var status common.Status
	allErrs := make([]error, 0)

read:
	for {
		select {
		case s, ok := <-responses:
			if !ok {
				break read
			}
			status = s
		case e := <-errs:
			allErrs = append(allErrs, e)
		}
	}

	// drain remaining errors
	for i := 0; i < len(errs); i++ {
		e := <-errs
		allErrs = append(allErrs, e)
	}
	// close errs channel since we have read all of them
	close(errs)
	return status, toError(allErrs)
}
155875
func (buffer *Buffer) awaitGenReply() error {
	var done osc.Message
	select {
	case done = <-buffer.client.doneChan:
	case err := <-buffer.client.errChan:
		return err
	}
	if len(done.Arguments) != 2 {
		return errors.New("expected two arguments to /done message")
	}
	addr, err := done.Arguments[0].ReadString()
	if err != nil {
		return err
	}
	// If reply address is not /b_gen, requeue the done event.
	if addr != bufferGenAddress {
		buffer.client.doneChan <- done
		return nil
	}
	bufnum, err := done.Arguments[1].ReadInt32()
	if err != nil {
		return err
	}

	// TODO: Don't error if we get a done message for a different buffer.
	// We should probably requeue this particular done message on doneChan.
	if bufnum != buffer.Num {
		buffer.client.doneChan <- done
	}
	return nil
}
262278
func newStreamWatcher(r io.ReadCloser) watchInterface {
	sw := &streamWatcher{
		r:       r,
		decoder: json.NewDecoder(r),
		result:  make(chan Event),
	}
	go sw.receive()
	return sw
}
1012790
func (c *Client) WaitContainer(id string) (int, error) {
	return c.waitContainer(id, doOptions{})
}
585072
func (sb *SafeBrowser) WaitUntilReady(ctx context.Context) error {
	if atomic.LoadUint32(&sb.closed) == 1 {
		return errClosed
	}
	select {
	case <-sb.db.Ready():
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-sb.done:
		return errClosed
	}
}
661183
func (node *ServerNode) ReceivedReply(reqID, bv uint64) {
	node.lock.Lock()
	defer node.lock.Unlock()

	now := node.clock.Now()
	node.recalcBLE(now)
	if bv > node.params.BufLimit {
		bv = node.params.BufLimit
	}
	sc, ok := node.pending[reqID]
	if !ok {
		return
	}
	delete(node.pending, reqID)
	cc := node.sumCost - sc
	newEstimate := uint64(0)
	if bv > cc {
		newEstimate = bv - cc
	}
	if newEstimate > node.bufEstimate {
		// Note: we never reduce the buffer estimate based on the reported value because
		// this can only happen because of the delayed delivery of the latest reply.
		// The lowest estimate based on the previous reply can still be considered valid.
		node.bufEstimate = newEstimate
	}

	node.bufRecharge = node.bufEstimate < node.params.BufLimit
	node.lastTime = now
	if node.log != nil {
		node.log.add(now, fmt.Sprintf("received  reqID=%d  bufEst=%d  reportedBv=%d  sumCost=%d  oldSumCost=%d", reqID, node.bufEstimate, bv, node.sumCost, sc))
	}
}
212862
func (c *Client) awaitBufAllocReply(buf *Buffer) error {
	var done osc.Message
	select {
	case done = <-c.doneChan:
	case err := <-c.errChan:
		return err
	}
	// error if this message was not an ack of /b_alloc
	if numargs := len(done.Arguments); numargs != 2 {
		return errors.Errorf("expected two arguments to /done message, got %d", numargs)
	}
	addr, err := done.Arguments[0].ReadString()
	if err != nil {
		return err
	}
	if addr != bufferAllocAddress {
		c.doneChan <- done

	}
	bufnum, err := done.Arguments[1].ReadInt32()
	if err != nil {
		return err
	}
	if bufnum != buf.Num {
		c.doneChan <- done
	}
	return nil
}
477845
func (_m *ClientInterface) WaitService(serviceIDs []string, state service.DesiredState, timeout time.Duration, recursive bool) error {
	ret := _m.Called(serviceIDs, state, timeout, recursive)

	var r0 error
	if rf, ok := ret.Get(0).(func([]string, service.DesiredState, time.Duration, bool) error); ok {
		r0 = rf(serviceIDs, state, timeout, recursive)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}
579438
func (d *DB) FetchWaitingCloseChannels() ([]*OpenChannel, error) {
	waitingClose, err := fetchChannels(d, false, true)
	if err != nil {
		return nil, err
	}
	pendingWaitingClose, err := fetchChannels(d, true, true)
	if err != nil {
		return nil, err
	}

	return append(waitingClose, pendingWaitingClose...), nil
}
1086488
func sendReply(w io.Writer, resp uint8, addr *AddrSpec) error {
	// Format the address
	var addrType uint8
	var addrBody []byte
	var addrPort uint16
	switch {
	case addr == nil:
		addrType = ipv4Address
		addrBody = []byte{0, 0, 0, 0}
		addrPort = 0

	case addr.FQDN != "":
		addrType = fqdnAddress
		addrBody = append([]byte{byte(len(addr.FQDN))}, addr.FQDN...)
		addrPort = uint16(addr.Port)

	case addr.IP.To4() != nil:
		addrType = ipv4Address
		addrBody = []byte(addr.IP.To4())
		addrPort = uint16(addr.Port)

	case addr.IP.To16() != nil:
		addrType = ipv6Address
		addrBody = []byte(addr.IP.To16())
		addrPort = uint16(addr.Port)

	default:
		return fmt.Errorf("Failed to format address: %v", addr)
	}

	// Format the message
	msg := make([]byte, 6+len(addrBody))
	msg[0] = socks5Version
	msg[1] = resp
	msg[2] = 0 // Reserved
	msg[3] = addrType
	copy(msg[4:], addrBody)
	msg[4+len(addrBody)] = byte(addrPort >> 8)
	msg[4+len(addrBody)+1] = byte(addrPort & 0xff)

	// Send the message
	_, err := w.Write(msg)
	return err
}
612593
func (sw *streamWatcher) Stop() {
	sw.Lock()
	defer sw.Unlock()
	if !sw.stopped {
		sw.stopped = true
		sw.r.Close()
	}
}
967104
func (c Client) Notify(serviceMethod string, args interface{}) error {
	req := &rpc.Request{
		ServiceMethod: serviceMethod,
		Seq:           seqNotify,
	}
	return c.codec.WriteRequest(req, args)
}