// Sets up the writer/producer for a channel using the given retry options.
428137
func setupProducerForChannel(retryOptions localconfig.Retry, haltChan chan struct{}, brokers []string, brokerConfig *sarama.Config, channel channel) (sarama.SyncProducer, error) {
	var err error
	var producer sarama.SyncProducer

	logger.Infof("[channel: %s] Setting up the producer for this channel...", channel.topic())

	retryMsg := "Connecting to the Kafka cluster"
	setupProducer := newRetryProcess(retryOptions, haltChan, channel, retryMsg, func() error {
		producer, err = sarama.NewSyncProducer(brokers, brokerConfig)
		return err
	})

	return producer, setupProducer.retry()
}
1147041
func SetOutput(out io.Writer) {
	std = New(out, std.Prefix(), std.Flags(), std.priority)
}
244560
func (p *Producer) backoff(failed int) {
	backoff := p.Backoff.Duration()

	p.Logger.WithFields(log.Fields{
		"failures": failed,
		"backoff":  backoff,
	}).Warn("put failures")

	time.Sleep(backoff)
}
1038139
func (md *RootMetadata) SetWriters(writers []keybase1.UserOrTeamID) {
	md.bareMd.SetWriters(writers)
}
1136511
func (p *Params) SetRetryOpts(value retry.Opts) {
	logger.Debugf("RetryOpts: %#+v", value)
	p.RetryOpts = value
}
102008
func (s *PipelineActivity) SetChannel(v *ChannelActivity) *PipelineActivity {
	s.Channel = v
	return s
}
758232
func (k *kafkaRequester) Setup() error {
	config := sarama.NewConfig()
	producer, err := sarama.NewAsyncProducer(k.urls, config)
	if err != nil {
		return err
	}

	consumer, err := sarama.NewConsumer(k.urls, nil)
	if err != nil {
		producer.Close()
		return err
	}
	partitionConsumer, err := consumer.ConsumePartition(k.topic, 0, sarama.OffsetNewest)
	if err != nil {
		producer.Close()
		consumer.Close()
		return err
	}

	k.producer = producer
	k.consumer = consumer
	k.partitionConsumer = partitionConsumer
	msg := make([]byte, k.payloadSize)
	for i := 0; i < k.payloadSize; i++ {
		msg[i] = 'A' + uint8(rand.Intn(26))
	}
	k.msg = &sarama.ProducerMessage{
		Topic: k.topic,
		Value: sarama.ByteEncoder(msg),
	}

	return nil
}
206552
func (c *Client) SetRetryCount(count int) *Client {
	c.RetryCount = count
	return c
}
650492
func (w *BroadcastWriter) Write(p []byte) (n int, err error) {
	for _, writer := range w.writers {
		n, err = writer.Write(p)
		if err != nil {
			return
		}
		if n != len(p) {
			err = io.ErrShortWrite
			return
		}
	}
	return len(p), nil
}
948810
func SetErrFile(w fdWriter) {
	glog.Infof("Setting ErrFile to fd %d...", w.Fd())
	errFile = w
	useColor = wantsColor(w.Fd())
}
1190245
func Retry(err error, message string, args ...interface{}) error {
	return WrapWithMessage(&RetryError{
		Message: fmt.Sprintf(message, args...),
		Err:     err,
	}, message, args...)
}
717772
func (s *PutEventStreamInput) SetWriteEventStream(v *WriteEventStream) *PutEventStreamInput {
	s.WriteEventStream = v
	return s
}
644065
func (w *Writer) writeAndRetryWithPriority(p Priority, s string) (int, error) {
	conn := w.getConn()
	if conn != nil {
		if n, err := w.write(conn, p, s); err == nil {
			return n, err
		}
	}

	var err error
	if conn, err = w.connect(); err != nil {
		return 0, err
	}
	return w.write(conn, p, s)
}
901422
func (s *SegmentDemographics) SetChannel(v *SetDimension) *SegmentDemographics {
	s.Channel = v
	return s
}
846217
func (prod *BufferedProducer) Configure(conf PluginConfigReader) {
	prod.onPrepareStop = prod.DefaultDrain
	prod.onStop = prod.DefaultClose
	prod.messages = NewMessageQueue(int(conf.GetInt("Channel", 8192)))
}
344247
func (nw *NotifyWorker) setUp() NotifyChannel {
	watcher, err := nw.config.Handler.SetUp()
	if err != nil {
		nw.catacomb.Kill(err)
	}
	if watcher == nil {
		nw.catacomb.Kill(errors.New("handler returned nil watcher"))
	} else if err := nw.catacomb.Add(watcher); err != nil {
		nw.catacomb.Kill(err)
	} else {
		return watcher.Changes()
	}
	return nil
}
1038746
func (rs *retryingSink) write(events ...Event) error {
	if err := rs.sink.Write(events...); err != nil {
		rs.failure()
		return err
	}

	rs.reset()
	return nil
}