// Distribute write messages to given kafka topic, choosing next destination
// partition from internal cycle. All messages written within single Produce
// call are atomically written to the same destination.
195606
func (p *roundRobinProducer) Distribute(topic string, messages ...*proto.Message) (offset int64, err error) {
	p.mu.Lock()
	part := p.next
	p.next++
	if p.next >= p.partitions {
		p.next = 0
	}
	p.mu.Unlock()

	return p.producer.Produce(topic, int32(part), messages...)
}
9508
func Distribute(rs Publisher, ms ...Sender) Publisher {
	return rs.React(func(r Publisher, err error, data interface{}) {
		for _, mi := range ms {
			go func(rec Sender) {
				if err != nil {
					rec.SendError(err)
					return
				}
				rec.Send(data)
			}(mi)
		}
	}, true)
}
430432
func (c *Conn) hideWriteMessages(msgs ...Message) (int, error) {
	if len(msgs) == 0 {
		return 0, nil
	}

	// users may believe they can set the Topic and/or Partition
	// on the kafka message.
	for _, msg := range msgs {
		if msg.Topic != "" && msg.Topic != c.topic {
			return 0, errInvalidWriteTopic
		}
		if msg.Partition != 0 {
			return 0, errInvalidWritePartition
		}
	}

	n := 0
	for _, msg := range msgs {
		n += len(msg.Key) + len(msg.Value)
	}

	err := c.writeOperation(
		func(deadline time.Time, id int32) error {
			now := time.Now()
			deadline = adjustDeadlineForRTT(deadline, now, defaultRTT)
			return writeProduceRequestV2(
				&c.wbuf,
				id,
				c.clientID,
				c.topic,
				c.partition,
				deadlineToTimeout(deadline, now),
				int16(atomic.LoadInt32(&c.requiredAcks)),
				msgs...,
			)
		},
		func(deadline time.Time, size int) error {
			return expectZeroSize(readArrayWith(&c.rbuf, size, func(r *bufio.Reader, size int) (int, error) {
				// Skip the topic, we've produced the message to only one topic,
				// no need to waste resources loading it in memory.
				size, err := discardString(r, size)
				if err != nil {
					return size, err
				}

				// Read the list of partitions, there should be only one since
				// we've produced a message to a single partition.
				size, err = readArrayWith(r, size, func(r *bufio.Reader, size int) (int, error) {
					var p produceResponsePartitionV2
					size, err := p.readFrom(r, size)
					if err == nil && p.ErrorCode != 0 {
						err = Error(p.ErrorCode)
					}
					return size, err
				})
				if err != nil {
					return size, err
				}

				// The response is trailed by the throttle time, also skipping
				// since it's not interesting here.
				return discardInt32(r, size)
			}))
		},
	)

	if err != nil {
		n = 0
	}

	return n, err
}
326122
func (s *syncWriter) Write(p []byte) (n int, err error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.lw.Write(p)
}
1038497
func (wal *baseWAL) WriteSync(msg WALMessage) {
	if wal == nil {
		return
	}

	wal.Write(msg)
	if err := wal.FlushAndSync(); err != nil {
		panic(fmt.Sprintf("Error flushing consensus wal buf to file. Error: %v \n", err))
	}
}
1073512
func (p *protoManualSyncPublisherKafka) Put(key string, message proto.Message, opts ...datasync.PutOption) error {
	_, err := p.conn.sendSyncMessage(p.topic, p.partition, key, message, true)
	return err
}
795125
func (w *Producer) MultiPublish(topic string, body [][]byte) error {
	cmd, err := MultiPublish(topic, body)
	if err != nil {
		return err
	}
	return w.sendCommand(cmd)
}
1242559
func (c *Consumer) Partitions(topic string) ([]int32, error) {
	c.l.Lock()
	defer c.l.Unlock()

	if c.metadata == nil {
		c.t.Errorf("Unexpected call to Partitions. Initialize the mock's topic metadata with SetMetadata.")
		return nil, sarama.ErrOutOfBrokers
	}
	if c.metadata[topic] == nil {
		return nil, sarama.ErrUnknownTopicOrPartition
	}

	return c.metadata[topic], nil
}
46843
func (c *Conversation) Write(b []byte) (int, error) {
	n := len(b)
	return n, c.Send(b2s(b))
}
206406
func Write(w io.Writer, message io.Reader, order binary.ByteOrder) (i int, err error) {
	data, err := ioutil.ReadAll(message)
	if err != nil {
		return 0, err
	}

	header := make([]byte, messageSizeInBytes)
	order.PutUint32(header, uint32(len(data)))
	return w.Write(append(header, data...))
}
327094
func (b *redisBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
	v, err := b.opts.Codec.Marshal(msg)
	if err != nil {
		return err
	}

	conn := b.pool.Get()
	_, err = redis.Int(conn.Do("PUBLISH", topic, v))
	conn.Close()

	return err
}
1159217
func (mw multiWriter) Write(p []byte) (int, error) {
	done := make(chan result, len(mw))
	for _, w := range mw {
		go send(w, p, done)
	}
	endResult := result{n: len(p)}
	for _ = range mw {
		res := <-done
		if res.err != nil && (endResult.err == nil || res.n < endResult.n) {
			endResult = res
		}
	}
	return endResult.n, endResult.err
}
1311322
func (h *Hash) Write(msg []byte) (int, error) {
	if h.done {
		return 0, errWriteAfterSum
	}
	return h.hash.Write(msg)
}
1017484
func Write(client Client, writer Writer, msg message.Msg) (message.Msg, error) {
	return sessionFunc(client, writer.Write(msg))
}
125286
func (p *bytesManualSyncPublisherKafka) Put(key string, data []byte) error {
	_, err := p.conn.SendSyncMessageToPartition(p.topic, p.partition, sarama.StringEncoder(key), sarama.ByteEncoder(data))
	return err
}
1200497
func (ref *SyncProducer) SendMsgByte(topic string, key []byte, msg []byte) (*ProducerMessage, error) {
	// generate a key if none supplied (used by hash partitioner)
	ref.WithFields(logging.Fields{"key": key, "msg": msg}).Debug("Sending")

	if key == nil || len(key) == 0 {
		md5Sum := fmt.Sprintf("%x", md5.Sum(msg))
		return ref.SendMsgToPartition(topic, ref.Partition, sarama.ByteEncoder(md5Sum), sarama.ByteEncoder(msg))
	}
	return ref.SendMsgToPartition(topic, ref.Partition, sarama.ByteEncoder(key), sarama.ByteEncoder(msg))
}
532866
func (c *CyclePDB) Write(towrite []boltcycle.KvPair) (err error) {
	atomic.AddInt64(&c.stats.TotalWriteCount, int64(len(towrite)))

	return c.blobs.latest().Update(func(tx *bolt.Tx) error {
		bucket := tx.Bucket(c.bucketTimesIn)
		if bucket == nil {
			return errorUnableToFindRootBucket
		}

		for i := 0; err == nil && i < len(towrite); i++ {
			p := towrite[i]
			err = bucket.Put(p.Key, p.Value)
		}
		return err
	})
}