var b bytes.Buffer
	gzWriter := gzip.NewWriter(&b)

	header := struct {
		Version string `json:"version"`
	}{Version: healthInfo.Version}

	enc := json.NewEncoder(gzWriter)
	if e := enc.Encode(header); e != nil {
		internalLogIf(ctx, fmt.Errorf("Could not encode health info header: %w", e))
		return nil
	}

	if e := enc.Encode(healthInfo); e != nil {

				}
			}
			atomic.StoreUint64(&ps.types, uint64(remainTypes))
			atomic.AddInt32(&ps.numSubscribers, -1)
		}()

		// Read from subChT and write to subCh
		var buf bytes.Buffer
		enc := json.NewEncoder(&buf)
		for {
			select {
			case <-doneCh:
				return
			case v, ok := <-subChT:
				if !ok {
					return
				}
				buf.Reset()
				err := enc.Encode(v)
				if err != nil {

func (r *Record) Raw() (sql.SelectObjectFormat, any) {
	return r.SelectFormat, r.KVS
}

// WriteJSON - encodes to JSON data.
func (r *Record) WriteJSON(writer io.Writer) error {
	return json.NewEncoder(writer).Encode(r.KVS)
}

// Replace the underlying buffer of json data.
func (r *Record) Replace(k any) error {
	v, ok := k.(jstream.KVS)
	if !ok {

func (store *QueueStore[I]) multiWrite(key Key, items []I) (err error) {
	buf := bytebufferpool.Get()
	defer bytebufferpool.Put(buf)

	enc := jsoniter.ConfigCompatibleWithStandardLibrary.NewEncoder(buf)

	for i := range items {
		if err = enc.Encode(items[i]); err != nil {
			return err
		}
	}

	path := filepath.Join(store.directory, key.String())
	if key.Compress {

func (w *Writer) Close() error {
	w.pw.CloseWithError(nil)

	if w.f != nil {
		if err := w.closeCurrentFile(); err != nil {
			return err
		}
	}

	return nil
}

var stdErrEnc = json.NewEncoder(os.Stderr)

func (w *Writer) listen() {
	for {
		var r io.Reader = w.pr
		if w.opts.MaximumFileSize > 0 {
			r = io.LimitReader(w.pr, w.opts.MaximumFileSize)
		}
		if _, err := io.Copy(w.f, r); err != nil {

		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
		return
	}

	w.Header().Set(xhttp.ContentType, string(mimeJSON))

	enc := json.NewEncoder(w)
	stats := globalReplicationStats.Load().getLatestReplicationStats(bucket)
	bwRpt := globalNotificationSys.GetBandwidthReports(ctx, bucket)
	bwMap := bwRpt.BucketStats
	for arn, st := range stats.ReplicationStats.Stats {

// keys, so we need to use this technique to merge the
// keys.
func sanitizePolicy(r io.Reader) (io.Reader, error) {
	var buf bytes.Buffer
	e := json.NewEncoder(&buf)
	d := jstream.NewDecoder(r, 0).ObjectAsKVS().MaxDepth(10)
	sset := set.NewStringSet()
	for mv := range d.Stream() {
		if mv.ValueType == jstream.Object {
			// This is a JSON object type (that preserves key order)

		contentType := r.Header.Get("Content-Type")
		if contentType == "application/json" {
			w.Header().Set("Content-Type", "application/json")
			jsonEncoder := json.NewEncoder(w)
			jsonEncoder.Encode(metrics)
			return
		}

		// If not JSON, return plain text. We format it as a markdown table for
		// readability.
		w.Header().Set("Content-Type", "text/plain")

				}
				versions[idx] = version{
					Idx:      idx,
					Header:   b,
					Metadata: buf.Bytes(),
				}
				return nil
			})
			if e != nil {
				return nil, e
			}
			enc := json.NewEncoder(buf)
			if e := enc.Encode(struct {
				Versions []version
			}{Versions: versions}); e != nil {
				return nil, e
			}
			data = b
		default:

			Name:     fmt.Sprintf("test-item-%d", i),
			Property: "property",
		})
	}

	buf := bytebufferpool.Get()
	defer bytebufferpool.Put(buf)

	enc := jsoniter.ConfigCompatibleWithStandardLibrary.NewEncoder(buf)
	for i := range items {
		if err = enc.Encode(items[i]); err != nil {
			t.Fatal(err)
		}
	}

	if _, err := store.PutMultiple(items); err != nil {
		t.Fatalf("failed to put multiple; %v", err)