// Close disables Append, calls flush and waits for this call to finish.
// Timeout is passed to WaitForFlush.
339028
func (batch *MessageBatch) Close(assemble AssemblyFunc, timeout time.Duration) {
	atomic.StoreInt32(batch.closed, 1)
	batch.Flush(assemble)
	batch.WaitForFlush(timeout)
}
471387
func (s *server) Close() error {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	if s.closed {
		return nil
	}
	return s.closeWithMutex()
}
753940
func (fl *File) Close() error {
	if fl.fo != nil {
		if err := fl.fo.Close(); err != nil {
			return err
		}
		fl.fo = nil
	}

	if fl.fe != nil {
		return fl.fe.Close()
	}

	return nil
}
977979
func (w *WriteOnCloser) Close() error {
	if !w.hasWritten {
		_, err := w.Write(nil)
		if err != nil {
			return err
		}
	}
	if closer, ok := w.Writer.(io.Closer); ok {
		return closer.Close()
	}
	return nil
}
694778
func (s *WriterSender) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if err := s.writer.Flush(); err != nil {
		return err
	}

	s.Send(message.NewBytesMessage(s.priority, bytes.TrimRightFunc(s.buffer.Bytes(), unicode.IsSpace)))
	s.buffer.Reset()
	s.writer.Reset(s.buffer)
	return nil
}
1289178
func (tp *BytePool) Close() {
	tp.Drain()
	if tp.drainTicker != nil {
		tp.drainTicker.Stop()
		tp.drainTicker = nil
	}
}
238869
func (s *Server) Close(timeout time.Duration) {
	if timeout <= 0 {
		for _, server := range s.servers {
			if err := server.Close(); err != nil {
				s.Logger.Warnf("web > Server [%s] shutdown failed: %s", server.Addr, err)
			} else {
				s.Logger.Infof("web > Server [%s] shutdown successfully", server.Addr)
			}
		}
	} else {
		ctx, cancel := scontext.WithTimeout(scontext.Background(), timeout)
		defer cancel()

		var g sync.WaitGroup
		g.Add(len(s.servers))
		for _, server := range s.servers {
			go func(server *http.Server) {
				if err := server.Shutdown(ctx); err != nil {
					s.Logger.Warnf("web > Server [%s] shutdown failed: %s", server.Addr, err)
				} else {
					s.Logger.Infof("web > Server [%s] shutdown gracefully", server.Addr)
				}
				g.Done()
			}(server)
		}
		g.Wait()
	}
}
311618
func (m *Memory) Close() error {
	m.cancel()
	m.Lock()
	defer m.Unlock()
	m.buf.Close()
	return nil
}
1227952
func (forwarder *BufferedForwarder) Close() error {
	forwarder.stopFunc()
	forwarder.threadsWaitingToDie.Wait()
	return forwarder.closeSender()
}
722469
func (pq *PrefixQueue) Close() error {
	pq.Lock()
	defer pq.Unlock()

	// Check if queue is already closed.
	if !pq.isOpen {
		return nil
	}

	// Close the LevelDB database.
	if err := pq.db.Close(); err != nil {
		return err
	}

	// Reset size and set isOpen to false.
	pq.size = 0
	pq.isOpen = false

	return nil
}
701301
func (m *FileMutex) Close() error {
	var ol syscall.Overlapped
	if err := unlockFileEx(m.fd, 0, 1, 0, &ol); err != nil {
		return err
	}
	return syscall.Close(m.fd)
}
213433
func (h *HttpOutputHandler) Close() error {
	if h == nil {
		return ErrNullReference
	}
	// Protect Close() so you can both evaluate the error and use it as a deferred cleanup
	h.mu.Lock()
	if h.already_closed {
		h.mu.Unlock() // unlock ASAP so other routines can also Close() in case of errors
		return errors.New("Already closed")
	}
	h.already_closed = true
	h.mu.Unlock()

	defer h.o.cfg.in_progress.Done()
	defer h.t.Close()

	switch h.mode {
	case StreamMode:
		NVM.Printf("[%s] HTTP_STREAM: Received all chunks and waiting end of streaming for %s", h.o.ClientId, h.o.FullName())
		close(h.pipeline)
		h.in_progress.Wait()
		h.check_chunk_status()
		if h.failure != nil {
			FUUU.Printf("[%s] HTTP_STREAM: Aborting streaming of %s due to failure (took %0.3fs)", h.o.ClientId, h.o.id, time.Since(h.o.fetch_start).Seconds())
			return errors.New("Failed to retrieve from the backend")
		}

	case BatchMode:
		h.check_chunk_status()

		if h.failure != nil {
			FUUU.Printf("[%s] HTTP_BUFFER: Returning NotFound due to failure for %s (took %0.3fs)", h.o.ClientId, h.o.id, time.Since(h.o.fetch_start).Seconds())
			http.NotFound(h.w, h.r)
			return errors.New("Failed to retrieve from the backend")
		}
		FYI.Printf("[%s] HTTP_BUFFER: Sending %s to client (took %0.3fs)", h.o.ClientId, h.o.id, time.Since(h.o.fetch_start).Seconds())
		h.send_output_headers()
		h.t.Seek(0, 0)
		io.Copy(h.w, h.t)
		return nil
	default:
		return errors.New("Unknown transfer mode")
	}
	return nil
}
1154041
func (f *File) Close() error {
	f.buf = nil
	f.dirIndex = 0

	return nil
}
335219
func (this *BufferedWriter) Close() error {
	close(this.queue)
	<-this.done
	return nil
}
804527
func (c *Compactor) Close() {
	c.mu.Lock()
	defer c.mu.Unlock()
	if !(c.snapshotsEnabled || c.compactionsEnabled) {
		return
	}
	c.snapshotsEnabled = false
	c.compactionsEnabled = false
	if c.compactionsInterrupt != nil {
		close(c.compactionsInterrupt)
	}
	if c.snapshotsInterrupt != nil {
		close(c.snapshotsInterrupt)
	}
}
666338
func (t *File) Close() error {
	t.closed = true
	return t.file.Close()
}
698011
func (h *Handler) Close() error {
	close(h.logHandlerChan)
	h.wg.Wait()
	return h.file.Close()
}