*
   * @throws IOException if an I/O error occurs while opening the stream
   */
  public abstract OutputStream openStream() throws IOException;

  /**
   * Opens a new buffered {@link OutputStream} for writing to this sink. The returned stream is not
   * required to be a {@link BufferedOutputStream} in order to allow implementations to simply

		t.Error("expected flush error, got nil")
	}
	if _, err := wr.ReadFrom(strings.NewReader("test2")); err == nil {
		t.Fatal("expected error, got nil")
	}
	if buffered := wr.Buffered(); buffered != wantBuffered {
		t.Fatalf("Buffered = %v; want %v", buffered, wantBuffered)
	}
}

func BenchmarkReaderCopyOptimal(b *testing.B) {
	// Optimal case is where the underlying reader implements io.WriterTo

   *
   * @throws IOException if an I/O error occurs while opening the writer
   */
  public abstract Writer openStream() throws IOException;

  /**
   * Opens a new buffered {@link Writer} for writing to this sink. The returned stream is not
   * required to be a {@link BufferedWriter} in order to allow implementations to simply delegate to

// does not exist in enough quorum nodes.
func refreshLock(ctx context.Context, ds *Dsync, id, source string, quorum int) (bool, error) {
	restClnts, _ := ds.GetLockers()

	// Create buffered channel of size equal to total number of nodes.
	ch := make(chan refreshResult, len(restClnts))
	var wg sync.WaitGroup

	args := LockArgs{
		UID: id,
	}

	for index, c := range restClnts {

}

// where we keep old *Readers
var readMsgpReaderPool = sync.Pool{New: func() interface{} { return &msgp.Reader{} }}

// mspNewReader returns a *Reader that reads from the provided reader.
// The reader will be buffered.
// Return with readMsgpReaderPoolPut when done.
func msgpNewReader(r io.Reader) *msgp.Reader {
	p := readMsgpReaderPool.Get().(*msgp.Reader)
	if p.R == nil {
		p.R = xbufio.NewReaderSize(r, 32<<10)
	} else {

	}
	return n, err
}

// diskHealthReader provides a wrapper that will update disk health on
// ctx, on every successful read.
// This should only be used directly at the os/syscall level,
// otherwise buffered operations may return false health checks.
func diskHealthReader(ctx context.Context, r io.Reader) io.Reader {
	// Check if context has a disk health check.
	tracker, ok := ctx.Value(healthDiskCtxKey{}).(*healthDiskCtxValue)

	}

	rulesMap := event.NewRulesMap(eventNames, pattern, event.TargetID{ID: mustGetUUID()})

	// Listen Publisher uses nonblocking publish and hence does not wait for slow subscribers.
	// Use buffered channel to take care of burst sends or slow w.Write()
	ch := make(chan event.Event, globalAPIConfig.getRequestsPoolCapacity())
	err := globalHTTPListen.Subscribe(mask, ch, ctx.Done(), func(ev event.Event) bool {

// It must be called before any data is written.
func (w *Writer) SetOffset(n int64) {
	if w.cw.count != 0 {
		panic("zip: SetOffset called after data was written")
	}
	w.cw.count = n
}

// Flush flushes any buffered data to the underlying writer.
// Calling Flush is not normally necessary; calling Close is sufficient.
func (w *Writer) Flush() error {
	return w.cw.w.(*bufio.Writer).Flush()
}

		return
	}
	setEventStreamHeaders(w)

	// Trace Publisher and peer-trace-client uses nonblocking send and hence does not wait for slow receivers.
	// Keep 100k buffered channel.
	// If receiver cannot keep up with that we drop events.
	traceCh := make(chan []byte, 100000)
	peers, _ := newPeerRestClients(globalEndpoints)

   *
   * @throws IOException if an I/O error occurs while opening the stream
   */
  public abstract InputStream openStream() throws IOException;

  /**
   * Opens a new buffered {@link InputStream} for reading from this source. The returned stream is
   * not required to be a {@link BufferedInputStream} in order to allow implementations to simply