// stop the replay go-routine.
	if h.store != nil {
		h.storeCtxCancel()
	}

	// Set logch to nil and close it.
	// This will block all Send operations,
	// and finish the existing ones.
	// All future ones will be discarded.
	h.logChMu.Lock()
	xioutil.SafeClose(h.logCh)
	h.logCh = nil
	h.logChMu.Unlock()

	if h.producer != nil {
		h.producer.Close()
		h.client.Close()
	}

					// Number of full blocks in skipped area
					seqNum = uint32(compOff / SSEDAREPackageBlockSize)
					// Skip this many inside a decrypted block to get to compression block start
					decryptSkip = compOff % SSEDAREPackageBlockSize
					// Skip this number of full blocks.
					skipEnc := compOff / SSEDAREPackageBlockSize
					skipEnc *= sseDAREEncPackageBlockSize

		}
	}()
}

// Trace - send http trace request to peer nodes
func (client *peerRESTClient) Trace(ctx context.Context, traceCh chan<- []byte, traceOpts madmin.ServiceTraceOpts) {
	go func() {
		for {
			// Blocks until context is canceled or an error occurs.
			client.doTrace(ctx, traceCh, traceOpts)
			select {
			case <-ctx.Done():
				return
			default:

					if err != nil {
						return tls.Certificate{}, fmt.Errorf("Unable to decrypt KES client private key as specified by the shell environment: %v", err)
					}
					keyBytes = pem.EncodeToMemory(&pem.Block{Type: privateKeyPEM.Type, Bytes: keyBytes})
				}
				certificate, err := tls.X509KeyPair(certBytes, keyBytes)
				if err != nil {

func (h *healSequence) stop() {
	h.cancelCtx()
}

// pushHealResultItem - pushes a heal result item for consumption in
// the heal-status API. It blocks if there are
// maxUnconsumedHealResultItems. When it blocks, the heal sequence
// routine is effectively paused - this happens when the server has
// accumulated the maximum number of heal records per heal
// sequence. When the client consumes further records, the heal

}

// WaitForConnect will block until a connection has been established or
// the context is canceled, in which case the context error is returned.
func (c *Connection) WaitForConnect(ctx context.Context) error {
	if debugPrint {
		fmt.Println(c.Local, "->", c.Remote, "WaitForConnect")
		defer fmt.Println(c.Local, "->", c.Remote, "WaitForConnect done")
	}
	c.connChange.L.Lock()

		dm2nd.Lock(id, source)

		// Release lock after 10 seconds
		go func() {
			time.Sleep(2 * testDrwMutexAcquireTimeout)
			// fmt.Println("Unlocking dm2")

			dm2nd.Unlock(context.Background())
		}()

		dm3rd.Lock(id, source)

		// fmt.Printf("3rd lock obtained after 1st & 2nd locks are released\n")
		time.Sleep(testDrwMutexRefreshCallTimeout)

		dm3rd.Unlock(context.Background())
	}()

	// There is flow control in both directions.
	StreamHandler struct {
		// Handle an incoming request. Initial payload is sent.
		// Additional input packets (if any) are streamed to request.
		// Upstream will block when request channel is full.
		// Response packets can be sent at any time.
		// Any non-nil error sent as response means no more responses are sent.
		Handle StreamHandlerFn

		// Subroute for handler.

		},
	}

	for i, c := range cases {
		t.Run(c.file, func(t *testing.T) {
			var err error
			var record sql.Record
			var result bytes.Buffer
			input := openTestFile(t, c.file)
			// Get above block size.
			input = append(input, input...)
			args := ReaderArgs{
				FileHeaderInfo:             use,
				RecordDelimiter:            c.recordDelimiter,
				FieldDelimiter:             c.fieldDelimiter,

	return pathJoin(bucketMetaPrefix, bucket, metacachePrefix, id)
}

// objectPath returns the object path of the cache.
func (o *listPathOptions) objectPath(block int) string {
	return pathJoin(metacachePrefixForID(o.Bucket, o.ID), "block-"+strconv.Itoa(block)+".s2")
}

func (o *listPathOptions) SetFilter() {
	switch {
	case metacacheSharePrefix:
		return
	case o.Prefix == o.BaseDir:
		// No additional prefix