125 lines
3.6 KiB
Go
125 lines
3.6 KiB
Go
package localserver
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type DispatchDataChannelCallBack = func(ctx context.Context, dc *DataChannel) (catched bool)
|
|
|
|
type ZoneRequestScheduler struct {
|
|
handlersPublishers []chan<- *ZoneRequest
|
|
handlersDataChannelDispatchCallbacks []DispatchDataChannelCallBack
|
|
}
|
|
|
|
type ZoneRequest struct {
|
|
ReqType string `json:"reqType"`
|
|
Payload map[string]any `json:"payload"`
|
|
From string `json:"from"`
|
|
}
|
|
|
|
type ZoneResponse struct {
|
|
Type string `json:"type"`
|
|
To string `json:"to"`
|
|
From string `json:"from"`
|
|
Payload map[string]any `json:"payload"`
|
|
}
|
|
|
|
type ZoneRequestHandler interface {
|
|
Init(ctx context.Context, authorizedMembers []string) (err error)
|
|
Subscribe(ctx context.Context, publisher <-chan *ZoneRequest) (reqChan chan *ZoneRequest, done chan struct{}, errCh chan error)
|
|
handleZoneRequest(ctx context.Context, req *ZoneRequest) (err error)
|
|
handleDataChannel(ctx context.Context, dc *DataChannel) (catched bool)
|
|
}
|
|
|
|
func NewZoneRequestScheduler(authorizedMembers []string, handlers ...ZoneRequestHandler) (zoneRequestScheduler *ZoneRequestScheduler, handlerErrCh chan error) {
|
|
zoneRequestScheduler = new(ZoneRequestScheduler)
|
|
zoneRequestScheduler.handlersPublishers = make([]chan<- *ZoneRequest, 0)
|
|
zoneRequestScheduler.handlersDataChannelDispatchCallbacks = make([]DispatchDataChannelCallBack, 0)
|
|
handlerErrCh = make(chan error)
|
|
reqChans := []chan *ZoneRequest{}
|
|
for _, handler := range handlers {
|
|
publisher := make(chan *ZoneRequest, 100)
|
|
zoneRequestScheduler.handlersPublishers = append(zoneRequestScheduler.handlersPublishers, publisher)
|
|
zoneRequestScheduler.handlersDataChannelDispatchCallbacks = append(zoneRequestScheduler.handlersDataChannelDispatchCallbacks, handler.handleDataChannel)
|
|
reqChan, done, errCh := handler.Subscribe(context.Background(), publisher)
|
|
go func(done <-chan struct{}, errCh <-chan error) {
|
|
for {
|
|
select {
|
|
case <-done:
|
|
return
|
|
case handlerErrCh <- <-errCh:
|
|
}
|
|
}
|
|
}(done, errCh)
|
|
reqChans = append(reqChans, reqChan)
|
|
}
|
|
|
|
for _, reqChan := range reqChans {
|
|
go func(reqChan <-chan *ZoneRequest) {
|
|
done, errCh := zoneRequestScheduler.Schedule(reqChan)
|
|
for {
|
|
select {
|
|
case <-done:
|
|
return
|
|
case e := <-errCh:
|
|
logger.Println("from there", e)
|
|
}
|
|
}
|
|
}(reqChan)
|
|
}
|
|
for i, handler := range handlers {
|
|
if ierr := handler.Init(context.Background(), authorizedMembers); ierr != nil {
|
|
logger.Println(ierr)
|
|
}
|
|
logger.Println("init done for handler", i)
|
|
}
|
|
return
|
|
}
|
|
|
|
func (zrs *ZoneRequestScheduler) Schedule(reqChan <-chan *ZoneRequest) (done chan struct{}, errCh chan error) {
|
|
done, errCh = make(chan struct{}), make(chan error)
|
|
for req := range reqChan {
|
|
go func(r *ZoneRequest) {
|
|
for _, publisher := range zrs.handlersPublishers {
|
|
go func(p chan<- *ZoneRequest) {
|
|
p <- r
|
|
}(publisher)
|
|
}
|
|
}(req)
|
|
}
|
|
done <- struct{}{}
|
|
return
|
|
}
|
|
|
|
func (zrs *ZoneRequestScheduler) DispatchDatachannel(ctx context.Context, dc *DataChannel) (catched bool) {
|
|
wg, lock := sync.WaitGroup{}, &sync.Mutex{}
|
|
timeoutCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
|
done := make(chan struct{})
|
|
defer cancel()
|
|
for _, dispatchCallback := range zrs.handlersDataChannelDispatchCallbacks {
|
|
wg.Add(1)
|
|
go func(w *sync.WaitGroup, dispatchCallback DispatchDataChannelCallBack) {
|
|
defer w.Done()
|
|
val := dispatchCallback(timeoutCtx, dc)
|
|
if val {
|
|
lock.Lock()
|
|
catched = true
|
|
lock.Unlock()
|
|
return
|
|
}
|
|
}(&wg, dispatchCallback)
|
|
}
|
|
go func() {
|
|
wg.Wait()
|
|
done <- struct{}{}
|
|
}()
|
|
select {
|
|
case <-done:
|
|
return
|
|
case <-timeoutCtx.Done():
|
|
return false
|
|
}
|
|
}
|