var buf bytes.Buffer
	e := json.NewEncoder(&buf)
	d := jstream.NewDecoder(r, 0).ObjectAsKVS()
	sset := set.NewStringSet()
	for mv := range d.Stream() {
		var kvs jstream.KVS
		if mv.ValueType == jstream.Object {
			// This is a JSON object type (that preserves key order)
			kvs = mv.Value.(jstream.KVS)
			for _, kv := range kvs {
				if sset.Contains(kv.Key) {

	// Ensure remote has the same disk ID.
	opts.DiskID = *client.diskID.Load()
	b, err := opts.MarshalMsg(grid.GetByteBuffer()[:0])
	if err != nil {
		return toStorageErr(err)
	}

	st, err := client.gridConn.NewStream(ctx, grid.HandlerWalkDir, b)
	if err != nil {
		return toStorageErr(err)
	}
	return toStorageErr(st.Results(func(in []byte) error {
		_, err := wr.Write(in)
		return err
	}))
}

	<-gotCall
	remote.debugMsg(debugKillInbound)
	local.debugMsg(debugKillInbound)
	<-gotResp

	// Must reconnect
	errFatal(remoteConn.WaitForConnect(context.Background()))

	stream, err := remoteConn.NewStream(context.Background(), handlerTest2, []byte(testPayload))
	errFatal(err)
	go func() {
		for resp := range stream.responses {
			t.Log("Resp:", resp, err)
		}
		gotResp <- struct{}{}
	}()

	<-gotCall

	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/klauspost/compress/zip"
	"github.com/minio/madmin-go/v3"
	"github.com/minio/madmin-go/v3/estream"
	"github.com/minio/minio-go/v7/pkg/set"
	"github.com/minio/minio/internal/dsync"
	"github.com/minio/minio/internal/grid"
	"github.com/minio/minio/internal/handlers"
	xhttp "github.com/minio/minio/internal/http"

		}
		return nil, io.EOF
	}

	var kvs jstream.KVS
	if v.ValueType == jstream.Object {
		// This is a JSON object type (that preserves key
		// order)
		kvs = v.Value.(jstream.KVS)
	} else {
		// To be AWS S3 compatible Select for JSON needs to
		// output non-object JSON as single column value
		// i.e. a map with `_1` as key and value as the
		// non-object.
		kvs = jstream.KVS{jstream.KV{Key: "_1", Value: v.Value}}
	}

func (r *Record) Replace(k interface{}) error {
	v, ok := k.(jstream.KVS)
	if !ok {
		return fmt.Errorf("cannot replace internal data in json record with type %T", k)
	}
	r.KVS = v
	return nil
}

// NewRecord - creates new empty JSON record.
func NewRecord(f sql.SelectObjectFormat) *Record {
	return &Record{
		KVS:          jstream.KVS{},
		SelectFormat: f,
	}
}

		return nil, errDataSource(errors.New("path not supported"))
	}
	switch rec := rawVal.(type) {
	case jstream.KVS:
		txedRec, _, err := jsonpathEval(e.selectAST.From.Table.PathExpr[1:], rec)
		if err != nil {
			return nil, err
		}

		var kvs jstream.KVS
		switch v := txedRec.(type) {
		case jstream.KVS:
			kvs = v

		case []interface{}:
			recs := make([]*Record, len(v))
			for i, val := range v {

github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k=
github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8=
github.com/bcicen/jstream v1.0.1 h1:BXY7Cu4rdmc0rhyTVyT3UkxAiX3bnLpKLas9btbH5ck=
github.com/bcicen/jstream v1.0.1/go.mod h1:9ielPxqFry7Y4Tg3j4BfjPocfJ3TbsRtXOAYXYmRuAQ=
github.com/beevik/ntp v1.3.1 h1:Y/srlT8L1yQr58kyPWFPZIxRL8ttx2SRIpVYJqZIlAM=

OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

================================================================

github.com/bcicen/jstream
https://github.com/bcicen/jstream
----------------------------------------------------------------
The MIT License (MIT)

Copyright (c) 2018 Bradley Cicenas