// flush records and retry failures if necessary.
182207
func (p *Producer) flush(records []*k.PutRecordsRequestEntry, reason string) {
	p.Logger.WithFields(log.Fields{
		"records": len(records),
		"reason":  reason,
	}).Info("flush")

	out, err := p.Client.PutRecords(&k.PutRecordsInput{
		StreamName: &p.StreamName,
		Records:    records,
	})

	if err != nil {
		p.Logger.WithError(err).Error("flush")
		p.backoff(len(records))
		p.flush(records, "error")
		return
	}

	failed := *out.FailedRecordCount
	if failed == 0 {
		p.Backoff.Reset()
		return
	}

	p.backoff(int(failed))
	p.flush(failures(records, out.Records), "retry")
}
865079
func (w *Writer) Flush() error {
	if w.buf.Len() == 0 {
		return nil
	}

	if err := w.clearLines(); err != nil {
		return err
	}
	w.lines = countLines(w.buf.String())

	if _, err := w.w.Write(w.buf.Bytes()); err != nil {
		return err
	}

	w.buf.Reset()
	return nil
}
993559
func (r *Reporter) Flush() {
	defer func() {
		if rec := recover(); rec != nil {
			handlePanic(rec)
		}
	}()

	for range time.Tick(r.interval) {
		if err := r.FlushOnce(); err != nil {
			log.Println(err)
		}
	}
}
995944
func (w *Writer) Flush() error {
	if _, err := w.w.Write(w.b); err != nil {
		return err
	}
	w.b = w.b[:0]
	return nil
}
1143762
func (bg *Writer) Flush() error {
	if bg.closed {
		return ErrClosed
	}
	if err := bg.Error(); err != nil {
		return err
	}

	if bg.active.next == 0 {
		return nil
	}

	var c *compressor
	c, bg.active = bg.active, <-bg.waiting
	bg.queue <- c
	bg.qwg.Add(1)
	go c.writeBlock()

	return bg.Error()
}
1281991
func (b *BulkIndexer) Flush() {
	b.mu.Lock()
	if b.docCt > 0 {
		b.send(b.buf)
	}
	b.mu.Unlock()
}
153679
func (s *KVStore) Flush(ctx context.Context) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, b := range s.buckets {
		b.btree.Clear(false)
	}
}
688199
func (p *BulkProcessor) Flush() error {
	p.statsMu.Lock()
	p.stats.Flushed++
	p.statsMu.Unlock()

	for _, w := range p.workers {
		w.flushC <- struct{}{}
		<-w.flushAckC // wait for completion
	}
	return nil
}
373926
func (w *compressWriter) Flush() {
	if w.encoder != nil {
		w.encoder.Flush()
	}
	if w, ok := w.ResponseWriter.(http.Flusher); ok {
		w.Flush()
	}
}
1162943
func (rw *responseWriter) Flush() {
	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	}
}
1090660
func (w *responseWriter) Flush() {
	if flusher, ok := w.Flusher(); ok {
		flusher.Flush()
	}
}
1331637
func (d *diagnosticsCollector) Flush() error {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.metrics["Uptime"] = (time.Now().Unix() - d.startTime)
	buf, err := d.encode()
	if err != nil {
		return errors.Wrap(err, "encoding")
	}
	req, err := http.NewRequest("POST", d.host, bytes.NewReader(buf))
	if err != nil {
		return errors.Wrap(err, "making new request")
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := d.client.Do(req)
	if err != nil {
		return errors.Wrap(err, "posting")
	}
	// Intentionally ignoring response body, as user does not need to be notified of error.
	defer resp.Body.Close()
	return nil
}
781017
func (w *Writer) Flush() error {
	w.seq++
	w.writePending()
	if w.err != nil {
		return w.err
	}
	if w.f != nil {
		w.err = w.f.Flush()
		return w.err
	}
	return nil
}
249648
func (w *responseWriter) Flush() {
	w.WriteHeaderNow()
	w.ResponseWriter.(http.Flusher).Flush()
}
129627
func (ctx *hookContext) Flush(process string, ctxErr error) (err error) {
	return ctx.recorder.Close()
}
1008789
func (b *batchProducer) Flush(timeout time.Duration, sendStats bool) (int, int, error) {
	b.Stop()

	timer := time.NewTimer(timeout)
	if timeout == 0 {
		timer.Stop()
	}

	timedOut := false
	sent := 0

loop:
	for len(b.records) > 0 {
		select {
		case <-timer.C:
			timedOut = true
			break loop
		default:
			sent += b.sendBatch(MaxKinesisBatchSize)
		}
	}

	if !timedOut && sendStats {
		b.sendStats()
	}

	return sent, len(b.records), nil
}
1326128
func (c *conn) Flush() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.buf.Flush()
}