// UnlimitedRowsChannel buffers incomming message from bufferThis channel and forward them to
// returned channel.
// ONLY CREATED CHANNEL IS CLOSED HERE.
431241
func UnlimitedRowsChannel(bufferThis chan message, firstMessage message) chan []string {
	driverChannel := make(chan []string)
	rowList := list.New()

	rowList.PushBack(firstMessage.Value)

	go func() {
		for {
			// If nothing comes in and nothing is going out...
			if rowList.Len() == 0 && bufferThis == nil {
				close(driverChannel)
				return
			}

			// Create a copy of driverChannel because if there is nothing to send
			// We can disable the case in select with a nil channel and get a chance
			// to fetch new data on bufferThis channel
			driverChannelNullable := driverChannel
			var nextRow []string
			if rowList.Len() != 0 {
				nextRow = rowList.Front().Value.([]string)
			} else {
				driverChannelNullable = nil
			}

			select {
			// In case a new row is sent to driver
			case driverChannelNullable <- nextRow:
				if nextRow != nil {
					rowList.Remove(rowList.Front())
				}
				// In case we receive a new value to buffer from engine channel
			case newRow, ok := <-bufferThis:
				if !ok || newRow.Type == rowEndMessage {
					// Stop listening to bufferThis channel
					bufferThis = nil
					// If there is nothing more to listen and there is nothing in buffer, exit
					// close driverChannel so *Rows knows there is nothing more to read
					if rowList.Len() == 0 {
						if driverChannel != nil {
							close(driverChannel)
						}
						return
					}

					if driverChannel == nil {
						log.Critical("Unlimited: But there is nobody to read it, exiting")
						return
					}
				} else if newRow.Type == errMessage {
					log.Critical("Runtime error: %s", newRow.Value[0])
					if driverChannel != nil {
						close(driverChannel)
					}
					return
				} else {
					// Everything is ok, buffering new value
					rowList.PushBack(newRow.Value)
				}
			case exit := <-driverChannel:
				// this means driverChannel is closed
				// set driverChannel to nil so we don't try to close it again
				driverChannel = nil
				_ = exit
			}
		}
	}()

	return driverChannel
}
351027
func (l *channelLink) handleBatchFwdErrs(errChan chan error) {
	for {
		err, ok := <-errChan
		if !ok {
			// Err chan has been drained or switch is shutting
			// down.  Either way, return.
			return
		}

		if err == nil {
			continue
		}

		l.errorf("unhandled error while forwarding htlc packet over "+
			"htlcswitch: %v", err)
	}
}
1138672
func NewPayloadsBuffer(next uint64) PayloadsBuffer {
	return &PayloadsBufferImpl{
		buf:       make(map[uint64]*proto.Payload),
		readyChan: make(chan struct{}, 1),
		next:      next,
		logger:    util.GetLogger(util.StateLogger, ""),
	}
}
1127520
func NewBoundedBuffer(n int) Buffer {
	if n <= 0 {
		panic("size must be greater than 0")
	}

	return &buffer{ring: make([]*Info, n, n)}
}
526303
func ChannelHeader(env *cb.Envelope) (*cb.ChannelHeader, error) {
	envPayload, err := UnmarshalPayload(env.Payload)
	if err != nil {
		return nil, err
	}

	if envPayload.Header == nil {
		return nil, errors.New("header not set")
	}

	if envPayload.Header.ChannelHeader == nil {
		return nil, errors.New("channel header not set")
	}

	chdr, err := UnmarshalChannelHeader(envPayload.Header.ChannelHeader)
	if err != nil {
		return nil, errors.WithMessage(err, "error unmarshaling channel header")
	}

	return chdr, nil
}
971089
func (c *Client) QueryBuffer(num int32) (*Buffer, error) {
	if err := c.oscConn.Send(osc.Message{
		Address: bufferQueryAddress,
		Arguments: osc.Arguments{
			osc.Int(num),
		},
	}); err != nil {
		return nil, errors.Wrap(err, "sending buffer query message")
	}
	return c.awaitBufInfoReply()
}
633864
func DoByteBufStreamingRoundTrip(stream testpb.BenchmarkService_StreamingCallClient, reqSize, respSize int) error {
	out := make([]byte, reqSize)
	if err := stream.(grpc.ClientStream).SendMsg(&out); err != nil {
		return fmt.Errorf("/BenchmarkService/StreamingCall.(ClientStream).SendMsg(_) = %v, want <nil>", err)
	}
	var in []byte
	if err := stream.(grpc.ClientStream).RecvMsg(&in); err != nil {
		// EOF is a valid error here.
		if err == io.EOF {
			return nil
		}
		return fmt.Errorf("/BenchmarkService/StreamingCall.(ClientStream).RecvMsg(_) = %v, want <nil>", err)
	}
	return nil
}
280857
func NewWebsocketHub() *WebsocketHub {
	return &WebsocketHub{
		clients:          make(map[*hubSpoke]*client),
		Register:         make(chan *clientHubSpoke),
		Unregister:       make(chan *hubSpoke),
		HubRelay:         make(chan pstypes.HubMessage),
		bufferTickerChan: make(chan int, clientSignalSize),
		quitWSHandler:    make(chan struct{}),
		requestLimit:     MaxPayloadBytes, // 1 MB
	}
}
625661
func (c *channel) WriteExtended(data []byte, extendedCode uint32) (n int, err error) {
	if c.sentEOF {
		return 0, io.EOF
	}
	// 1 byte message type, 4 bytes remoteId, 4 bytes data length
	opCode := byte(msgChannelData)
	headerLength := uint32(9)
	if extendedCode > 0 {
		headerLength += 4
		opCode = msgChannelExtendedData
	}

	c.writeMu.Lock()
	packet := c.packetPool[extendedCode]
	// We don't remove the buffer from packetPool, so
	// WriteExtended calls from different goroutines will be
	// flagged as errors by the race detector.
	c.writeMu.Unlock()

	for len(data) > 0 {
		space := min(c.maxRemotePayload, len(data))
		if space, err = c.remoteWin.reserve(space); err != nil {
			return n, err
		}
		if want := headerLength + space; uint32(cap(packet)) < want {
			packet = make([]byte, want)
		} else {
			packet = packet[:want]
		}

		todo := data[:space]

		packet[0] = opCode
		binary.BigEndian.PutUint32(packet[1:], c.remoteId)
		if extendedCode > 0 {
			binary.BigEndian.PutUint32(packet[5:], uint32(extendedCode))
		}
		binary.BigEndian.PutUint32(packet[headerLength-4:], uint32(len(todo)))
		copy(packet[headerLength:], todo)
		if err = c.writePacket(packet); err != nil {
			return n, err
		}

		n += len(todo)
		data = data[len(todo):]
	}

	c.writeMu.Lock()
	c.packetPool[extendedCode] = packet
	c.writeMu.Unlock()

	return n, err
}
179412
func UnmarshalChannelHeaderOrPanic(bytes []byte) *cb.ChannelHeader {
	chdr, err := UnmarshalChannelHeader(bytes)
	if err != nil {
		panic(err)
	}
	return chdr
}
62565
func (vse *Engine) StreamRows(ctx context.Context, query string, lastpk []sqltypes.Value, send func(*binlogdatapb.VStreamRowsResponse) 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)
	log.Infof("Streaming rows for query %s, lastpk: %s", query, lastpk)

	// Create stream and add it to the map.
	rowStreamer, idx, err := func() (*rowStreamer, int, error) {
		vse.mu.Lock()
		defer vse.mu.Unlock()
		if !vse.isOpen {
			return nil, 0, errors.New("VStreamer is not open")
		}
		rowStreamer := newRowStreamer(ctx, vse.cp, vse.se, query, lastpk, vse.kschema, send)
		idx := vse.streamIdx
		vse.rowStreamers[idx] = rowStreamer
		vse.streamIdx++
		// Now that we've added the stream, increment wg.
		// This must be done before releasing the lock.
		vse.wg.Add(1)
		return rowStreamer, 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.rowStreamers, idx)
		vse.wg.Done()
	}()

	// No lock is held while streaming, but wg is incremented.
	return rowStreamer.Stream()
}
370678
func SendIPMessageToChannel(client *TwilioIPMessagingClient, serviceSid string, channelSid string, from string, body string) (*IPMessage, error) {
	var message *IPMessage

	params := url.Values{}
	params.Set("Body", body)
	if from != "" {
		params.Set("From", from)
	}

	res, err := client.post(params, "/Services/"+serviceSid+"/Channels/"+channelSid+"/Messages.json")

	if err != nil {
		return message, err
	}

	message = new(IPMessage)
	err = json.Unmarshal(res, message)

	return message, err
}
1365049
func (s *Session) ChannelMessageSend(channelID string, content string) (*Message, error) {
	return s.ChannelMessageSendComplex(channelID, &MessageSend{
		Content: content,
	})
}
655984
func (opts *Options) SetMaxWriteBufferNumber(value int) {
	C.rocksdb_options_set_max_write_buffer_number(opts.c, C.int(value))
}
452537
func (p *ChannelPackager) AckAddHtlcs(tx *bbolt.Tx, addRefs ...AddRef) error {
	if len(addRefs) == 0 {
		return nil
	}

	fwdPkgBkt := tx.Bucket(fwdPackagesKey)
	if fwdPkgBkt == nil {
		return ErrCorruptedFwdPkg
	}

	sourceKey := makeLogKey(p.source.ToUint64())
	sourceBkt := fwdPkgBkt.Bucket(sourceKey[:])
	if sourceBkt == nil {
		return ErrCorruptedFwdPkg
	}

	// Organize the forward references such that we just get a single slice
	// of indexes for each unique height.
	heightDiffs := make(map[uint64][]uint16)
	for _, addRef := range addRefs {
		heightDiffs[addRef.Height] = append(
			heightDiffs[addRef.Height],
			addRef.Index,
		)
	}

	// Load each height bucket once and remove all acked htlcs at that
	// height.
	for height, indexes := range heightDiffs {
		err := ackAddHtlcsAtHeight(sourceBkt, height, indexes)
		if err != nil {
			return err
		}
	}

	return nil
}
1367060
func (wsh *WebsocketHub) periodicTxBufferSend() {
	ticker := time.NewTicker(bufferTickerInterval * time.Second)
	for {
		select {
		case <-ticker.C:
			wsh.SetTimeToSendTxBuffer(true)
		case sig := <-wsh.bufferTickerChan:
			switch sig {
			case tickerSigReset:
				ticker.Stop()
				ticker = time.NewTicker(bufferTickerInterval * time.Second)
			case tickerSigStop:
				close(wsh.bufferTickerChan)
				return
			}
		}
	}
}
574242
func NewRingBuffer(size int) RingBuffer {
	rb := &ringBuffer{}
	rb.start = &valueLink{}
	rb.end = rb.start
	if size < 2 {
		size = 2
	}
	for i := 0; i < size-1; i++ {
		link := &valueLink{}
		rb.end.next = link
		rb.end = link
	}
	rb.end.next = rb.start
	return rb
}