// azureWriteCloser is a io.WriteCloser that manages the azure connection pipe and when Close is called
// it blocks until all data is flushed to azure via a background go routine call to uploadMultiPart.
425824
func newAzureWriteCloser(ctx context.Context, f *FS, obj *object) io.WriteCloser {
	pr, pw := io.Pipe()
	bw := bufio.NewWriter(pw)

	g, _ := errgroup.WithContext(ctx)

	g.Go(func() error {
		// Upload the file to azure.
		// Do a multipart upload
		err := f.uploadMultiPart(obj, pr)
		if err != nil {
			gou.Warnf("could not upload %v", err)
			return err
		}
		return nil
	})

	return azureWriteCloser{
		pr, pw, bw, g,
	}
}
1135404
func PipeWrite(w io.WriteCloser, pipelines ...WritePipeline) (io.WriteCloser, error) {
	pipes := []io.WriteCloser{w}
	for _, pipeline := range pipelines {
		w, err := pipeline(pipes[len(pipes)-1].(io.WriteCloser))
		if err != nil {
			return nil, err
		}
		pipes = append(pipes, w)
	}
	return &writePipe{
		pipes: pipes,
		pipe:  pipes[len(pipes)-1],
	}, nil
}
938302
func NewWriteCloseInformer(wc io.WriteCloser) (io.WriteCloser, <-chan struct{}) {
	close := make(chan struct{})
	return &writeCloseInformer{
		close: close,
		wc:    wc,
	}, close
}
916567
func (f *win32MessageBytePipe) CloseWrite() error {
	if f.writeClosed {
		return errPipeWriteClosed
	}
	err := f.win32File.Flush()
	if err != nil {
		return err
	}
	_, err = f.win32File.Write(nil)
	if err != nil {
		return err
	}
	f.writeClosed = true
	return nil
}
1187231
func (rw *GCS) NewWriter(id string) (io.WriteCloser, error) {
	ctx := context.Background()
	obj := rw.bucket.Object(rw.Prefixed.Name(id))
	return obj.NewWriter(ctx), nil
}
929079
func (c Conn) CloseWrite() error {
	if cw, ok := c.Wire.(interface {
		CloseWrite() error
	}); ok {
		return cw.CloseWrite()
	}
	return errors.New("pinned: underlying connection does not support CloseWrite")
}
219626
func (c *Cache) WriteCloser(id string) (ItemInfo, io.WriteCloser, error) {
	id = cleanID(id)
	c.nlocker.Lock(id)

	info := ItemInfo{Name: id}

	f, err := helpers.OpenFileForWriting(c.Fs, id)
	if err != nil {
		c.nlocker.Unlock(id)
		return info, nil, err
	}

	return info, &lockedFile{
		File:   f,
		unlock: func() { c.nlocker.Unlock(id) },
	}, nil
}
29136
func CompressStream(dest io.Writer, compression Compression) (io.WriteCloser, error) {
	switch compression {
	case Uncompressed:
		return &writeCloserWrapper{dest, nil}, nil
	case Gzip:
		return gzip.NewWriter(dest), nil
	default:
		return nil, fmt.Errorf("unsupported compression format %s", (&compression).Extension())
	}
}
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
}
1280781
func (fs *FileSorter) ClosePartCollectors() (err error) {
	for _, mo := range fs.mapOuts {
		if e := mo.writer.Close(); e != nil {
			err = e
		}
	}
	return err
}
146337
func (b *Buffer) ReadCloser() io.ReadCloser {
	ret := &readCloser{0, append(b.bufs, b.Buf)}

	b.bufs = nil
	b.toPool = nil
	b.Buf = nil

	return ret
}
1285874
func (wf *WriteFlusher) Close() error {
	wf.closeLock.Lock()
	defer wf.closeLock.Unlock()

	select {
	case <-wf.closed:
		return errWriteFlusherClosed
	default:
		close(wf.closed)
	}
	return nil
}
1124511
func (w *Writer) ReadCloser() (io.ReadCloser, error) {
	if w.Error != nil {
		return nil, w.Error
	}

	return w.Buffer.ReadCloser(), nil
}
215325
func (s *S3Provider) Writer(path ...string) (io.WriteCloser, error) {
	s3FilePath := strings.Join(path, "/")
	s3, err := storage.SafeCreateS3Bucket(s.S3Domain, s.BucketName, s.AccessKeyID, s.SecretAccessKey)

	if err != nil {
		return nil, err
	}
	return s3.NewWriter(s3FilePath)
}
153112
func (p *IOThrottlerPool) AddWriter(writer io.WriteCloser) (io.WriteCloser, error) {
	// Register with the pool
	err := twiddleClientCount(p, 1)
	if err != nil {
		return nil, err
	}

	return &throttledWriteCloser{writer, p}, nil
}
458671
func (f *FileWriter) Close() error {
	if f.closed {
		return io.ErrClosedPipe
	}

	var lastBlock *hdfs.ExtendedBlockProto
	if f.blockWriter != nil {
		lastBlock = f.blockWriter.Block.GetB()

		// Close the blockWriter, flushing any buffered packets.
		err := f.finalizeBlock()
		if err != nil {
			return err
		}
	}

	completeReq := &hdfs.CompleteRequestProto{
		Src:        proto.String(f.name),
		ClientName: proto.String(f.client.namenode.ClientName),
		Last:       lastBlock,
	}
	completeResp := &hdfs.CompleteResponseProto{}

	err := f.client.namenode.Execute("complete", completeReq, completeResp)
	if err != nil {
		return &os.PathError{"create", f.name, err}
	}

	return nil
}
449876
func NewWriteCloserOnError(w io.WriteCloser, notify func(error)) io.WriteCloser {
	return NewWriteCloser(NewWriterOnError(w, notify), w)
}