// Put publishes a message into kafka
487468
func (p *protoAsyncPublisherKafka) Put(key string, message proto.Message, opts ...datasync.PutOption) error {
	return p.conn.sendAsyncMessage(p.topic, DefPartition, key, message, false, nil, p.succCallback, p.errCallback)
}
1238615
func (c *Client) Put(endpoint string, body []byte) (*RawResponse, error) {

	logger.Debug("put %s, %s, [%s]", endpoint, body, c.Cluster.Leader)
	p := endpoint

	req := NewRawRequest("PUT", p, body, nil)
	resp, err := c.SendRequest(req)

	if err != nil {
		return nil, err
	}

	return resp, nil
}
1370759
func (c *Client) Put(key string, data []byte, opts ...datasync.PutOption) error {
	newEntry := &decoder.FileDataEntry{Key: key, Value: data}
	statusDataEntries := c.db.GetDataForFile(c.statusPath)
	// Add/update data
	var updated bool
	for _, statusDataEntry := range statusDataEntries {
		if statusDataEntry.Key == key {
			statusDataEntry.Value = data
			updated = true
			break
		}
	}
	if !updated {
		statusDataEntries = append(statusDataEntries, newEntry)
	}
	c.db.Add(c.statusPath, newEntry)
	// Encode and write
	stFileEntries, err := c.statusDecoder.Encode(statusDataEntries)
	if err != nil {
		return errors.Errorf("failed to write status to fileDB: unable to encode status file %s: %v", c.statusPath, err)
	}
	err = c.fsHandler.WriteFile(c.statusPath, stFileEntries)
	if err != nil {
		return errors.Errorf("failed to write status %s to fileDB: %v", c.statusPath, err)
	}
	return nil
}
439509
func (d *Failstore) Put(k ds.Key, val []byte) error {
	err := d.errfunc("put")
	if err != nil {
		return err
	}

	return d.child.Put(k, val)
}
211383
func (adapter *Adapter) Put(key string, data proto.Message) error {
	if adapter.Publisher != nil {
		return adapter.Publisher.Put(key, data)
	}

	return nil
}
1013429
func (c *Client) Put(key string, value []byte, flags uint64) error {
	_, err := c.putCAS(key, value, flags, 0, false)
	return err
}
37813
func (store *MemoryStore) Put(key string, message packets.ControlPacket) {
	store.Lock()
	defer store.Unlock()
	if !store.opened {
		ERROR.Println(STR, "Trying to use memory store, but not open")
		return
	}
	store.messages[key] = message
}
355176
func (bk *Bucket) Put(k, v []byte) error {
	return bk.db.Update(func(tx *bolt.Tx) error {
		return tx.Bucket(bk.Name).Put(k, v)
	})
}
812874
func (r *Resource) Put(options ...interface{}) (*Resource, error) {
	if len(options) > 0 {
		r.Payload = r.SetPayload(options[0])
	}
	return r.do("PUT")
}
365981
func (c *Consumer) commit(offsets []TopicPartition) (committedOffsets []TopicPartition, err error) {
	var rkqu *C.rd_kafka_queue_t

	rkqu = C.rd_kafka_queue_new(c.handle.rk)
	defer C.rd_kafka_queue_destroy(rkqu)

	var coffsets *C.rd_kafka_topic_partition_list_t
	if offsets != nil {
		coffsets = newCPartsFromTopicPartitions(offsets)
		defer C.rd_kafka_topic_partition_list_destroy(coffsets)
	}

	cErr := C.rd_kafka_commit_queue(c.handle.rk, coffsets, rkqu, nil, nil)
	if cErr != C.RD_KAFKA_RESP_ERR_NO_ERROR {
		return nil, newError(cErr)
	}

	rkev := C.rd_kafka_queue_poll(rkqu, C.int(-1))
	if rkev == nil {
		// shouldn't happen
		return nil, newError(C.RD_KAFKA_RESP_ERR__DESTROY)
	}
	defer C.rd_kafka_event_destroy(rkev)

	if C.rd_kafka_event_type(rkev) != C.RD_KAFKA_EVENT_OFFSET_COMMIT {
		panic(fmt.Sprintf("Expected OFFSET_COMMIT, got %s",
			C.GoString(C.rd_kafka_event_name(rkev))))
	}

	cErr = C.rd_kafka_event_error(rkev)
	if cErr != C.RD_KAFKA_RESP_ERR_NO_ERROR {
		return nil, newErrorFromCString(cErr, C.rd_kafka_event_error_string(rkev))
	}

	cRetoffsets := C.rd_kafka_event_topic_partition_list(rkev)
	if cRetoffsets == nil {
		// no offsets, no error
		return nil, nil
	}
	committedOffsets = newTopicPartitionsFromCparts(cRetoffsets)

	return committedOffsets, nil
}
1054432
func (s *MemoryStore) Put(key string, value interface{}) error {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	s.store[key] = value
	return nil
}
841074
func (pdb *BytesBrokerWatcherEtcd) Put(key string, data []byte, opts ...datasync.PutOption) error {
	return putInternal(pdb.Logger, pdb.kv, pdb.lessor, pdb.opTimeout, key, data, opts...)
}
494290
func (f *fsClient) Put(ctx context.Context, reader io.Reader, size int64, metadata map[string]string, progress io.Reader, sse encrypt.ServerSide) (int64, *probe.Error) {
	return f.put(reader, size, nil, progress)
}
498693
func (store *MemoryStore) Put(e event.Event) error {
	store.Lock()
	defer store.Unlock()
	if store.eC == store.limit {
		return errLimitExceeded
	}
	key, kErr := getNewUUID()
	if kErr != nil {
		return kErr
	}
	store.events[key] = e
	store.eC++
	return nil
}
771771
func (b *Bucket) Put(key []byte, value []byte) error {
	err := b.bucket.Put(key, value)
	if err == bolt.ErrTxNotWritable {
		return kv.ErrTxNotWritable
	}
	return err
}
834528
func (b *Backend) Put(ctx context.Context, entry *physical.Entry) (retErr error) {
	defer metrics.MeasureSince(metricPut, time.Now())

	// Pooling
	b.permitPool.Acquire()
	defer b.permitPool.Release()

	// Insert
	w := b.client.Bucket(b.bucket).Object(entry.Key).NewWriter(ctx)
	w.ChunkSize = b.chunkSize
	md5Array := md5.Sum(entry.Value)
	w.MD5 = md5Array[:]
	defer func() {
		closeErr := w.Close()
		if closeErr != nil {
			retErr = multierror.Append(retErr, errwrap.Wrapf("error closing connection: {{err}}", closeErr))
		}
	}()

	if _, err := w.Write(entry.Value); err != nil {
		return errwrap.Wrapf("failed to put data: {{err}}", err)
	}
	return nil
}
383900
func (m *Mercury) Put(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
	return m.cliv3.Put(ctx, m.parseKey(key), val, opts...)
}