func NewStreamLayer(r io.Reader, format string, c Cipher) (Layer, error) {
	return NewStreamLayerContext(context.Background(), r, format, c)
}
NewStreamLayer create new stream layer, see the NewStreamLayerContext
package onion
import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
)
var (
	decLock  sync.RWMutex
	decoders = map[string]Decoder{
		"json": &jsonDecoder{},
	}
)
type Cipher interface {
	Decrypt(io.Reader) ([]byte, error)
}
type Decoder interface {
	Decode(context.Context, io.Reader) (map[string]interface{}, error)
}
type jsonDecoder struct {
}
func decrypt(c Cipher, r io.Reader) (io.Reader, error) {
	if c == nil {
		return r, nil
	}
	b, err := c.Decrypt(r)
	if err != nil {
		return nil, err
	}
	return bytes.NewReader(b), nil
}
func (jd *jsonDecoder) Decode(_ context.Context, r io.Reader) (map[string]interface{}, error) {
	var data map[string]interface{}
	if err := json.NewDecoder(r).Decode(&data); err != nil {
		return nil, err
	}
	return data, nil
}
func RegisterDecoder(dec Decoder, formats ...string) {
	decLock.Lock()
	defer decLock.Unlock()
	for _, format := range formats {
		format := strings.ToLower(format)
		_, alreadyExists := decoders[format]
		if alreadyExists {
			log.Fatalf("decoder for format %q is already registered: you can have only one", format)
		}
		decoders[format] = dec
	}
}
func GetDecoder(format string) Decoder {
	decLock.RLock()
	defer decLock.RUnlock()
	return decoders[strings.ToLower(format)]
}
type streamLayer struct {
	c      chan map[string]interface{}
	cipher Cipher
}
func (sl *streamLayer) Load() map[string]interface{} {
	return <-sl.c
}
func (sl *streamLayer) Watch() <-chan map[string]interface{} {
	return sl.c
}
func (sl *streamLayer) Reload(ctx context.Context, r io.Reader, format string) error {
	dec := GetDecoder(format)
	if dec == nil {
		return fmt.Errorf("format %q is not registered", format)
	}
	dr, err := decrypt(sl.cipher, r)
	if err != nil {
		return err
	}
	data, err := dec.Decode(ctx, dr)
	if err != nil {
		return err
	}
	go func() {
		select {
		case sl.c <- data:
		case <-ctx.Done():
		}
	}()
	return nil
}
func NewStreamLayerContext(ctx context.Context, r io.Reader, format string, c Cipher) (Layer, error) {
	if r == nil {
		return nil, fmt.Errorf("nil stream")
	}
	sl := &streamLayer{
		c:      make(chan map[string]interface{}),
		cipher: c,
	}
	err := sl.Reload(ctx, r, format)
	if err != nil {
		return nil, err
	}
	return sl, nil
}