func (m *SubscriptionManager) Create(
	ctx context.Context,
	key string,
	subscriber Subscriber,
	c chan<- interface{},
) error {
	if len(key) == 0 {
		panic("key must be set")
	}
	if subscriber == nil {
		panic("subscriber must not be nil")
	}
	return m.master.Create(ctx, key, createSubscriptionRequest{
		Subscriber: subscriber,
		C:          c,
	})
}
Create a new subscription identified by the
specified key
package eth
import (
	"context"
	"fmt"
	"math/big"
	"sync"
	ethereum "github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/oasislabs/oasis-gateway/concurrent"
	"github.com/oasislabs/oasis-gateway/log"
)
type EthSubscription struct {
	sub ethereum.Subscription
	err chan error
}
func (s *EthSubscription) Unsubscribe() {
	s.sub.Unsubscribe()
}
func (s *EthSubscription) Err() <-chan error {
	return s.err
}
type LogSubscriber struct {
	lock        sync.Mutex
	FilterQuery ethereum.FilterQuery
	BlockNumber uint64
	Index       uint
}
func (s *LogSubscriber) createSubscription(
	ctx context.Context,
	client Client,
	clog chan<- types.Log,
) (ethereum.Subscription, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	return client.SubscribeFilterLogs(ctx, s.FilterQuery, clog)
}
func (s *LogSubscriber) Subscribe(
	ctx context.Context,
	client Client,
	c chan<- interface{},
) (ethereum.Subscription, error) {
	cerr := make(chan error)
	clog := make(chan types.Log, 64)
	sub, err := s.createSubscription(ctx, client, clog)
	if err != nil {
		return nil, err
	}
	go func() {
		defer func() {
			s.lock.Lock()
			defer s.lock.Unlock()
			s.FilterQuery.FromBlock = big.NewInt(0).SetUint64(s.BlockNumber)
			close(cerr)
		}()
		for {
			select {
			case <-ctx.Done():
				return
			case ev, ok := <-clog:
				if !ok {
					return
				}
				if ev.BlockNumber < s.BlockNumber ||
					(ev.BlockNumber == s.BlockNumber && ev.Index < s.Index) {
					continue
				}
				s.lock.Lock()
				s.BlockNumber = ev.BlockNumber
				s.Index = ev.Index
				s.lock.Unlock()
				c <- ev
			case err, ok := <-sub.Err():
				if !ok {
					return
				}
				cerr <- err
				return
			}
		}
	}()
	return &EthSubscription{sub: sub, err: cerr}, nil
}
type Subscriber interface {
	Subscribe(context.Context, Client, chan<- interface{}) (ethereum.Subscription, error)
}
type SubscriptionEndEvent struct {
	Key string
	Error error
}
type SubscriptionProps struct {
	Logger log.Logger
	Client Client
	URL string
	Key string
	Subscriber Subscriber
	C chan<- interface{}
}
type Subscription struct {
	logger     log.Logger
	client     Client
	sub        ethereum.Subscription
	subscriber Subscriber
	url        string
	key        string
	c          chan<- interface{}
}
func NewSubscription(props SubscriptionProps) *Subscription {
	if props.Logger == nil {
		panic("logger cannot be nil")
	}
	if props.Client == nil {
		panic("client must be set")
	}
	if props.Subscriber == nil {
		panic("subscriber must be set")
	}
	if props.C == nil {
		panic("receiving channel must be set")
	}
	s := &Subscription{
		logger:     props.Logger.ForClass("eth", "Subscription"),
		client:     props.Client,
		url:        props.URL,
		subscriber: props.Subscriber,
		key:        props.Key,
		c:          props.C,
	}
	return s
}
func (s *Subscription) subscribe(ctx context.Context) error {
	sub, err := s.subscriber.Subscribe(ctx, s.client, s.c)
	if err != nil {
		return err
	}
	s.sub = sub.(ethereum.Subscription)
	return nil
}
func (s *Subscription) Key() string {
	return s.key
}
func (s *Subscription) Unsubscribe() {
	s.sub.Unsubscribe()
}
func (s *Subscription) handle(ctx context.Context, ev concurrent.WorkerEvent) (interface{}, error) {
	switch ev := ev.(type) {
	case concurrent.RequestWorkerEvent:
		panic("no requests should be issued to the subscription")
	case concurrent.ErrorWorkerEvent:
		err := s.handleError(ctx, ev)
		return nil, err
	default:
		panic("received unexpected event type")
	}
}
func (s *Subscription) handleError(ctx context.Context, ev concurrent.ErrorWorkerEvent) error {
	s.logger.Debug(ctx, "subscription failed, recreating", log.MapFields{
		"call_type": "CurrentSubscriptionFailure",
		"err":       ev.Error.Error(),
	})
	return s.subscribe(ctx)
}
type createSubscriptionRequest struct {
	C          chan<- interface{}
	Subscriber Subscriber
}
type SubscriptionManagerProps struct {
	Context context.Context
	Logger log.Logger
	Client Client
}
type SubscriptionManager struct {
	ctx    context.Context
	logger log.Logger
	client Client
	master *concurrent.Master
}
func NewSubscriptionManager(
	props SubscriptionManagerProps,
) *SubscriptionManager {
	m := SubscriptionManager{
		ctx:    props.Context,
		logger: props.Logger.ForClass("eth", "SubscriptionManager"),
		client: props.Client,
	}
	m.master = concurrent.NewMaster(concurrent.MasterProps{
		MasterHandler: concurrent.MasterHandlerFunc(m.handle),
	})
	if err := m.master.Start(props.Context); err != nil {
		panic(fmt.Sprintf("failed to start loop %s", err.Error()))
	}
	return &m
}
func (m *SubscriptionManager) handle(ctx context.Context, ev concurrent.MasterEvent) error {
	switch ev := ev.(type) {
	case concurrent.CreateWorkerEvent:
		return m.create(ctx, ev)
	case concurrent.DestroyWorkerEvent:
		return m.destroy(ev)
	default:
		panic("received unknown request")
	}
}
func (m *SubscriptionManager) create(ctx context.Context, ev concurrent.CreateWorkerEvent) error {
	req := ev.Value.(createSubscriptionRequest)
	sub := NewSubscription(SubscriptionProps{
		Logger:     m.logger,
		Client:     m.client,
		Key:        ev.Key,
		Subscriber: req.Subscriber,
		C:          req.C,
	})
	if err := sub.subscribe(m.ctx); err != nil {
		return err
	}
	ev.Props.ErrC = sub.sub.Err()
	ev.Props.WorkerHandler = concurrent.WorkerHandlerFunc(sub.handle)
	ev.Props.UserData = sub
	return nil
}
func (m *SubscriptionManager) destroy(ev concurrent.DestroyWorkerEvent) error {
	sub := ev.Worker.UserData.(*Subscription)
	sub.Unsubscribe()
	return nil
}