// If the request context is canceled, the stream will no longer process requests.
	// Requests sent cannot be used any further by the called.
	Requests chan<- []byte

	muxID uint64
	ctx   context.Context
}

// Send a payload to the remote server.
func (s *Stream) Send(b []byte) error {
	if s.Requests == nil {
		return errors.New("stream does not accept requests")
	}
	select {

 */
@ExperimentalOkHttpApi
interface Stream {
  val requestBody: BufferedSource
  val responseBody: BufferedSink

  /**
   * Terminate the stream so that no further data is transmitted or received. Note that
   * [requestBody] may return data after this call; that is the buffered data received before this
   * stream was canceled.
   *

	"github.com/tinylib/msgp/msgp"
	"github.com/valyala/bytebufferpool"
)

// metadata stream format:
//
// The stream is s2 compressed.
// https://github.com/klauspost/compress/tree/master/s2#s2-compression
// This ensures integrity and reduces the size typically by at least 50%.
//
// All stream elements are msgpack encoded.
//
// 1 Integer, metacacheStreamVersion of the writer.

	}
	extracted := false
	for {
		stream, err := sr.NextStream()
		if err != nil {
			if err == io.EOF {
				if extracted {
					return nil
				}
				return errors.New("no data found on stream")
			}
			if errors.Is(err, estream.ErrNoKey) {
				if stream.Name == "inspect.zip" {
					return errors.New("incorrect private key")
				}
				if err := stream.Skip(); err != nil {

					continue
				}
				dst, ok := r.kvDstPool.Get().([]jstream.KVS)
				if !ok {
					dst = make([]jstream.KVS, 0, 1000)
				}

				d := jstream.NewDecoder(bytes.NewBuffer(in.input), 0).ObjectAsKVS().MaxDepth(100)
				stream := d.Stream()
				all := dst[:0]
				for mv := range stream {
					var kvs jstream.KVS
					if mv.ValueType == jstream.Object {
						// This is a JSON object type (that preserves key

```go
decoder := jstream.NewDecoder(f, 2).EmitKV() // enable KV streaming at a depth level of 2
```

```
jstream.KV{desc RGB}
jstream.KV{colors [red green blue]}
jstream.KV{desc CMYK}
jstream.KV{colors [cyan magenta yellow black]}
```

## Installing 

```bash
go get github.com/bcicen/jstream
```

## Commandline

		}
		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}}
	}

	"reflect"
	"testing"

	"github.com/alecthomas/participle"
	"github.com/minio/minio/internal/s3select/jstream"
)

func getJSONStructs(b []byte) ([]interface{}, error) {
	dec := jstream.NewDecoder(bytes.NewBuffer(b), 0).ObjectAsKVS().MaxDepth(100)
	var result []interface{}
	for parsedVal := range dec.Stream() {
		result = append(result, parsedVal.Value)
	}
	if err := dec.Err(); err != nil {
		return nil, err
	}

package jstream

import (
	"bytes"
	"testing"
)

func mkReader(s string) *bytes.Reader { return bytes.NewReader([]byte(s)) }

func TestDecoderSimple(t *testing.T) {
	var (
		counter int
		mv      *MetaValue
		body    = `[{"bio":"bada bing bada boom","id":1,"name":"Charles","falseVal":false}]`
	)

	decoder := NewDecoder(mkReader(body), 1)

	for mv = range decoder.Stream() {
		counter++

	var buf bytes.Buffer
	e := json.NewEncoder(&buf)
	d := jstream.NewDecoder(r, 0).ObjectAsKVS().MaxDepth(10)
	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) {