// Close closes the underlying kafka consumer.
172055
func (s *Source) Close() error {
	err := s.consumer.Close()
	return errors.Wrap(err, "closing kafka consumer")
}
716803
func (cnsmr *consumer) Close() error {
	if cnsmr.ws == nil {
		return errors.New("connection does not exist")
	}

	return cnsmr.ws.Close()
}
345795
func (q *Queue) Close() error {
	// Make sure all watchers have been closed to avoid a deadlock when
	// closing the broadcaster.
	q.mu.Lock()
	for _, cancelFunc := range q.cancelFuncs {
		cancelFunc()
	}
	q.cancelFuncs = make(map[events.Sink]func())
	q.mu.Unlock()

	return q.broadcast.Close()
}
564874
func (e *UnionExec) Close() error {
	close(e.finished)
	e.childrenResults = nil
	if e.resultPool != nil {
		for range e.resultPool {
		}
	}
	e.resourcePools = nil
	return e.baseExecutor.Close()
}
153059
func (m *PoolMetadata) Close() error {
	if err := m.db.Close(); err != nil && err != bolt.ErrDatabaseNotOpen {
		return err
	}

	return nil
}
91426
func (writer *SocketWriter) Close() {
	writer.lock.Lock()
	defer writer.lock.Unlock()
	if writer.closed {
		return
	}

	writer.writer.Close()
	writer.writer = nil
	writer.closed = true
}
935442
func (c *GRPCClient) Close() error {
	c.broker.Close()
	c.controller.Shutdown(c.doneCtx, &plugin.Empty{})
	return c.Conn.Close()
}
638505
func (w *Watcher) Close() error {
	if w.isClosed() {
		return nil
	}

	// Send 'close' signal to goroutine, and set the Watcher to closed.
	close(w.done)

	// Wake up goroutine
	w.poller.wake()

	// Wait for goroutine to close
	<-w.doneResp

	return nil
}
146243
func (c *Consumer) Close() error {
	c.l.Lock()
	defer c.l.Unlock()

	for _, partitions := range c.partitionConsumers {
		for _, partitionConsumer := range partitions {
			_ = partitionConsumer.Close()
		}
	}

	return nil
}
156257
func (c *Client) Close() error {
	c.mutex.Lock()
	if c.shutdown || c.closing {
		c.mutex.Unlock()
		return ErrShutdown
	}
	c.closing = true
	c.mutex.Unlock()
	return c.codec.Close()
}
937382
func (conn *gRPCQueryClient) Close(ctx context.Context) error {
	conn.mu.Lock()
	defer conn.mu.Unlock()
	if conn.cc == nil {
		return nil
	}

	cc := conn.cc
	conn.cc = nil
	return cc.Close()
}
571943
func (p *PoolConn) Close() error {
	p.mu.RLock()
	defer p.mu.RUnlock()

	if p.unusable {
		if p.Conn != nil {
			return p.Conn.Close()
		}
		return nil
	}
	return p.c.put(p.Conn)
}
390104
func (s *SocketHub) Close() {
	if s.closed {
		return
	}
	s.closed = true
	close(s.closer)
}
768781
func (bl *bufListener) Close() error {
	if !atomic.CompareAndSwapUint64(&bl.closed, 0, 1) {
		return nil
	}
	close(bl.conns)
	return nil
}
63585
func (t *Topic) Close() {
	oldQueueLen := C.int(0x7FFFFFFF)
	queueLen := C.rd_kafka_outq_len(t.client.handle)

	// Wait as long as we're flushing
	for queueLen > 0 && queueLen < oldQueueLen {
		C.rd_kafka_poll(t.client.handle, 1000)
		oldQueueLen = queueLen
		queueLen = C.rd_kafka_outq_len(t.client.handle)
	}

	if queueLen > 0 {
		Log.Printf("%d messages have been lost as the internal queue for %s could not be flushed", queueLen, t.name)
	}
	C.rd_kafka_topic_destroy(t.handle)
}
1227952
func (forwarder *BufferedForwarder) Close() error {
	forwarder.stopFunc()
	forwarder.threadsWaitingToDie.Wait()
	return forwarder.closeSender()
}
1018392
func (c *StreamConnection) Close() {
	c.lock.Lock()
	defer c.lock.Unlock()

	if c.Closed() {
		return
	}

	logger.Debug("Closing stream....")
	if err := c.stream.CloseSend(); err != nil {
		logger.Warnf("error closing GRPC stream: %s", err)
	}

	c.GRPCConnection.Close()
}