// Produce writes to a buffer that is sent to a given socket.
509823
func (prod *KafkaProducer) Produce(workers *sync.WaitGroup) {
	prod.tryConnect()
	prod.AddMainWorker(workers)
	prod.TickerMessageControlLoop(prod.produceMessage, prod.pollInterval, prod.poll)
}
995977
func (c *client) write(buf []byte) error {
	_, err := c.conn.Write(buf)
	return err
}
423629
func (g *GifGenerator) Generate(input string, output string) (err error) {
	// virtual terminal
	var state = terminal.State{}
	vt, err := terminal.Create(&state, ioutil.NopCloser(bytes.NewBuffer([]byte{})))
	if err != nil {
		return
	}
	defer vt.Close()
	vt.Resize(g.Col, g.Row)

	// play and capture
	var (
		images []*image.Paletted
		delays []int
	)
	err = g.TtyPlay(input, vt, func(diff int32) (err error) {
		delay := int(float64(diff)/g.Speed) / 10000
		if delay > 0 {
			var img *image.Paletted
			img, err = g.Capture(&state)
			if err != nil {
				return
			}
			images = append(images, img)
			delays = append(delays, delay)
		}
		return nil
	})
	if err != nil {
		return
	}

	// generate gif file
	outFile, err := os.Create(output)
	if err != nil {
		return
	}
	defer outFile.Close()
	opts := gif.GIF{
		Image: images,
		Delay: delays,
	}
	if g.NoLoop {
		opts.LoopCount = 1
	}
	err = gif.EncodeAll(outFile, &opts)
	if err != nil {
		return
	}
	return nil
}
1061821
func (p *Producer) loop() {
	buf := make([]*k.PutRecordsRequestEntry, 0, p.BufferSize)
	tick := time.NewTicker(p.FlushInterval)
	drain := false

	defer tick.Stop()
	defer close(p.done)

	for {
		select {
		case record := <-p.records:
			buf = append(buf, record)

			if len(buf) >= p.BufferSize {
				p.flush(buf, "buffer size")
				buf = nil
			}

			if drain && len(p.records) == 0 {
				p.Logger.Info("drained")
				return
			}
		case <-tick.C:
			if len(buf) > 0 {
				p.flush(buf, "interval")
				buf = nil
			}
		case <-p.done:
			drain = true
			if len(buf) > 0 {
				p.flush(buf, "stopping")
				buf = nil
			}

			if len(p.records) == 0 {
				return
			}
		}
	}
}
116492
func (prod *AwsKinesis) Produce(workers *sync.WaitGroup) {
	defer prod.WorkerDone()

	prod.AddMainWorker(workers)
	prod.initKinesisClient()
	prod.BatchMessageLoop(workers, prod.sendBatch)
}
830419
func (b *builderBuffer) Write(v []byte) {
	l := uint(len(v))
	if l > 0 {
		off := uint(len(*b))
		b.growCapacity(l)
		*b = (*b)[:off+l]
		copy((*b)[off:], v)
	}
}
842939
func (s *stream) transport() {
	defer s.teardown()

	bufsize := 256
	buf := new(bytes.Buffer)
	timer := time.NewTimer(0)
	if !timer.Stop() {
		select {
		case <-timer.C:
		default:
		}
	}

	// cleanup
	defer timer.Stop()

	// writeBuf writes the contents of buf through to the s.Writer.
	// done only when arrival time makes sense.
	drainBuf := func() error {
		if buf.Len() > 0 {
			_, err := s.write.Write(buf.Bytes())
			if err != nil {
				return err
			}
			buf.Reset()
		}
		return nil
	}

	// deliverOrWait is a helper func that processes
	// an incoming packet. it waits until the arrival time,
	// and then writes things out.
	deliverOrWait := func(o *transportObject) error {
		buffered := len(o.msg) + buf.Len()

		// Yes, we can end up extending a timer multiple times if we
		// keep on making small writes but that shouldn't be too much of an
		// issue. Fixing that would be painful.
		if !timer.Stop() {
			// FIXME: So, we *shouldn't* need to do this but we hang
			// here if we don't... Go bug?
			select {
			case <-timer.C:
			default:
			}
		}
		delay := o.arrivalTime.Sub(time.Now())
		if delay >= 0 {
			timer.Reset(delay)
		} else {
			timer.Reset(0)
		}

		if buffered >= bufsize {
			select {
			case <-timer.C:
			case <-s.reset:
				select {
				case s.reset <- struct{}{}:
				default:
				}
				return ErrReset
			}
			if err := drainBuf(); err != nil {
				return err
			}
			// write this message.
			_, err := s.write.Write(o.msg)
			if err != nil {
				return err
			}
		} else {
			buf.Write(o.msg)
		}
		return nil
	}

	for {
		// Reset takes precedent.
		select {
		case <-s.reset:
			s.writeErr = ErrReset
			return
		default:
		}

		select {
		case <-s.reset:
			s.writeErr = ErrReset
			return
		case <-s.close:
			if err := drainBuf(); err != nil {
				s.resetWith(err)
				return
			}
			s.writeErr = s.write.Close()
			if s.writeErr == nil {
				s.writeErr = ErrClosed
			}
			return
		case o := <-s.toDeliver:
			if err := deliverOrWait(o); err != nil {
				s.resetWith(err)
				return
			}
		case <-timer.C: // ok, due to write it out.
			if err := drainBuf(); err != nil {
				s.resetWith(err)
				return
			}
		}
	}
}
709599
func (p *ReadBuffer) Take() *buffer.Read {
	return p.pool.Take().(*buffer.Read)
}
486676
func (b *Buffer) Consume(n int) []byte {
	v, ok := b.consume(n)
	if !ok {
		return nil
	}
	return v
}
362114
func (i *tm) Generate() (string, error) {
	return strconv.Itoa(int(i.Now().Unix())), nil
}
586988
func (prod *Spooling) Produce(workers *sync.WaitGroup) {
	prod.AddMainWorker(workers)
	prod.spoolCheck = time.AfterFunc(prod.respoolDuration, prod.openExistingFiles)
	prod.TickerMessageControlLoop(prod.writeToFile, prod.batchTimeout, prod.writeBatchOnTimeOut)
}
572654
func (prod *StatsdMetrics) Produce(workers *sync.WaitGroup) {
	prod.AddMainWorker(workers)
	prod.TickerMessageControlLoop(prod.bufferMessage, prod.flushFrequency, prod.sendBatchOnTimeOut)
}
518739
func (p *WriteBuffer) Return(buf *buffer.Write) {
	p.pool.Return(buf)
}
1344443
func channelBatchProducer(p *Producer) {
	var buffered = make(map[string][]*Message)
	bufferedCnt := 0
	const batchSize int = 1000000
	totMsgCnt := 0
	totBatchCnt := 0

	for m := range p.produceChannel {
		buffered[*m.TopicPartition.Topic] = append(buffered[*m.TopicPartition.Topic], m)
		bufferedCnt++

	loop2:
		for true {
			select {
			case m, ok := <-p.produceChannel:
				if !ok {
					break loop2
				}
				if m == nil {
					panic("nil message received on ProduceChannel")
				}
				if m.TopicPartition.Topic == nil {
					panic(fmt.Sprintf("message without Topic received on ProduceChannel: %v", m))
				}
				buffered[*m.TopicPartition.Topic] = append(buffered[*m.TopicPartition.Topic], m)
				bufferedCnt++
				if bufferedCnt >= batchSize {
					break loop2
				}
			default:
				break loop2
			}
		}

		totBatchCnt++
		totMsgCnt += len(buffered)

		for topic, buffered2 := range buffered {
			err := p.produceBatch(topic, buffered2, C.RD_KAFKA_MSG_F_BLOCK)
			if err != nil {
				for _, m = range buffered2 {
					m.TopicPartition.Error = err
					p.events <- m
				}
			}
		}

		buffered = make(map[string][]*Message)
		bufferedCnt = 0
	}
	p.handle.terminatedChan <- "channelBatchProducer"
}
385804
func (p *PlaybackDevice) Write(buffer interface{}) (samples int, err error) {
	bufferType := reflect.TypeOf(buffer)
	if !(bufferType.Kind() == reflect.Array ||
		bufferType.Kind() == reflect.Slice) {
		return 0, errors.New("Write requires an array type")
	}

	sizeError := errors.New("Write requires a matching sample size")
	switch bufferType.Elem().Kind() {
	case reflect.Int8:
		if p.formatSampleSize() != 1 {
			return 0, sizeError
		}
	case reflect.Int16:
		if p.formatSampleSize() != 2 {
			return 0, sizeError
		}
	case reflect.Int32, reflect.Float32:
		if p.formatSampleSize() != 4 {
			return 0, sizeError
		}
	case reflect.Float64:
		if p.formatSampleSize() != 8 {
			return 0, sizeError
		}
	default:
		return 0, errors.New("Write does not support this format")
	}

	val := reflect.ValueOf(buffer)
	length := val.Len()
	sliceData := val.Slice(0, length)

	var frames = C.snd_pcm_uframes_t(length / p.Channels)
	bufPtr := unsafe.Pointer(sliceData.Index(0).Addr().Pointer())

	ret := C.snd_pcm_writei(p.h, bufPtr, frames)
	if ret == -C.EPIPE {
		C.snd_pcm_prepare(p.h)
		return 0, ErrUnderrun
	} else if ret < 0 {
		return 0, createError("write error", C.int(ret))
	}
	samples = int(ret) * p.Channels
	return
}
696138
func (prod *AwsCloudwatchLogs) Produce(workers *sync.WaitGroup) {
	prod.initService()
	if err := prod.create(); err != nil {
		prod.Logger.Errorf("could not create group:%q stream:%q error was: %q", prod.group, prod.stream, err)
	} else {
		prod.setToken()
	}
	prod.BatchMessageLoop(workers, prod.sendBatch)
}
830166
func Generate(raw string) (pwd string, salt string, err error) {
	return Default.Generate(raw)
}