Skip to content

Commit

Permalink
generalized event handler
Browse files Browse the repository at this point in the history
  • Loading branch information
DarienRaymond committed Nov 28, 2017
1 parent 973ce07 commit fd8db49
Show file tree
Hide file tree
Showing 16 changed files with 80 additions and 32 deletions.
2 changes: 1 addition & 1 deletion app/dispatcher/impl/default.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ func NewDefaultDispatcher(ctx context.Context, config *dispatcher.Config) (*Defa
return nil, newError("no space in context")
}
d := &DefaultDispatcher{}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
d.ohm = proxyman.OutboundHandlerManagerFromSpace(space)
if d.ohm == nil {
return newError("OutboundHandlerManager is not found in the space")
Expand Down
2 changes: 1 addition & 1 deletion app/dns/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ func NewCacheServer(ctx context.Context, config *dns.Config) (*CacheServer, erro
servers: make([]NameServer, len(config.NameServers)),
hosts: config.GetInternalHosts(),
}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
disp := dispatcher.FromSpace(space)
if disp == nil {
return newError("dispatcher is not found in the space")
Expand Down
4 changes: 4 additions & 0 deletions app/policy/manager/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,20 +37,24 @@ func global() policy.Policy {
}
}

// GetPolicy implements policy.Manager.
func (m *Instance) GetPolicy(level uint32) policy.Policy {
if p, ok := m.levels[level]; ok {
return *p
}
return global()
}

// Start implements app.Application.Start().
func (m *Instance) Start() error {
return nil
}

// Close implements app.Application.Close().
func (m *Instance) Close() {
}

// Interface implement app.Application.Interface().
func (m *Instance) Interface() interface{} {
return (*policy.Manager)(nil)
}
Expand Down
2 changes: 1 addition & 1 deletion app/proxyman/mux/mux.go
Original file line number Diff line number Diff line change
Expand Up @@ -274,7 +274,7 @@ type Server struct {
func NewServer(ctx context.Context) *Server {
s := &Server{}
space := app.SpaceFromContext(ctx)
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
d := dispatcher.FromSpace(space)
if d == nil {
return newError("no dispatcher in space")
Expand Down
2 changes: 1 addition & 1 deletion app/proxyman/outbound/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ func NewHandler(ctx context.Context, config *proxyman.OutboundHandlerConfig) (*H
if space == nil {
return nil, newError("no space in context")
}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
ohm := proxyman.OutboundHandlerManagerFromSpace(space)
if ohm == nil {
return newError("no OutboundManager in space")
Expand Down
2 changes: 1 addition & 1 deletion app/router/router.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ func NewRouter(ctx context.Context, config *Config) (*Router, error) {
rules: make([]Rule, len(config.Rule)),
}

space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
for idx, rule := range config.Rule {
r.rules[idx].Tag = rule.Tag
cond, err := rule.BuildCondition()
Expand Down
36 changes: 17 additions & 19 deletions app/space.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"reflect"

"v2ray.com/core/common"
"v2ray.com/core/common/event"
)

type Application interface {
Expand All @@ -13,8 +14,6 @@ type Application interface {
Close()
}

type InitializationCallback func() error

func CreateAppFromConfig(ctx context.Context, config interface{}) (Application, error) {
application, err := common.CreateObject(ctx, config)
if err != nil {
Expand All @@ -29,46 +28,45 @@ func CreateAppFromConfig(ctx context.Context, config interface{}) (Application,
}

// A Space contains all apps that may be available in a V2Ray runtime.
// Caller must check the availability of an app by calling HasXXX before getting its instance.
type Space interface {
event.Registry
GetApplication(appInterface interface{}) Application
AddApplication(application Application) error
Initialize() error
OnInitialize(InitializationCallback)
Start() error
Close()
}

const (
SpaceInitializing event.Event = iota
)

type spaceImpl struct {
initialized bool
event.Listener
cache map[reflect.Type]Application
appInit []InitializationCallback
initialized bool
}

func NewSpace() Space {
return &spaceImpl{
cache: make(map[reflect.Type]Application),
appInit: make([]InitializationCallback, 0, 32),
cache: make(map[reflect.Type]Application),
}
}

func (s *spaceImpl) OnInitialize(f InitializationCallback) {
if s.initialized {
f()
} else {
s.appInit = append(s.appInit, f)
func (s *spaceImpl) On(e event.Event, h event.Handler) {
if e == SpaceInitializing && s.initialized {
_ = h(nil) // Ignore error
return
}
s.Listener.On(e, h)
}

func (s *spaceImpl) Initialize() error {
for _, f := range s.appInit {
if err := f(); err != nil {
return err
}
if s.initialized {
return nil
}
s.appInit = nil
s.initialized = true
return nil
return s.Fire(SpaceInitializing, nil)
}

func (s *spaceImpl) GetApplication(appInterface interface{}) Application {
Expand Down
46 changes: 46 additions & 0 deletions common/event/event.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
package event

import "sync"

type Event uint16

type Handler func(data interface{}) error

type Registry interface {
On(Event, Handler)
}

type Listener struct {
sync.RWMutex
events map[Event][]Handler
}

func (l *Listener) On(e Event, h Handler) {
l.Lock()
defer l.Unlock()

if l.events == nil {
l.events = make(map[Event][]Handler)
}

handlers := l.events[e]
handlers = append(handlers, h)
l.events[e] = handlers
}

func (l *Listener) Fire(e Event, data interface{}) error {
l.RLock()
defer l.RUnlock()

if l.events == nil {
return nil
}

for _, h := range l.events[e] {
if err := h(data); err != nil {
return err
}
}

return nil
}
2 changes: 1 addition & 1 deletion proxy/dokodemo/dokodemo.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ func New(ctx context.Context, config *Config) (*DokodemoDoor, error) {
address: config.GetPredefinedAddress(),
port: net.Port(config.Port),
}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
pm := policy.FromSpace(space)
if pm == nil {
return newError("Policy not found in space.")
Expand Down
2 changes: 1 addition & 1 deletion proxy/freedom/freedom.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ func New(ctx context.Context, config *Config) (*Handler, error) {
timeout: config.Timeout,
destOverride: config.DestinationOverride,
}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
if config.DomainStrategy == Config_USE_IP {
f.dns = dns.FromSpace(space)
if f.dns == nil {
Expand Down
2 changes: 1 addition & 1 deletion proxy/http/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ func NewServer(ctx context.Context, config *ServerConfig) (*Server, error) {
s := &Server{
config: config,
}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
pm := policy.FromSpace(space)
if pm == nil {
return newError("Policy not found in space.")
Expand Down
2 changes: 1 addition & 1 deletion proxy/shadowsocks/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) {
if space == nil {
return nil, newError("Space not found.")
}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
pm := policy.FromSpace(space)
if pm == nil {
return newError("Policy not found in space.")
Expand Down
2 changes: 1 addition & 1 deletion proxy/shadowsocks/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ func NewServer(ctx context.Context, config *ServerConfig) (*Server, error) {
account: account,
}

space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
pm := policy.FromSpace(space)
if pm == nil {
return newError("Policy not found in space.")
Expand Down
2 changes: 1 addition & 1 deletion proxy/socks/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ func NewServer(ctx context.Context, config *ServerConfig) (*Server, error) {
s := &Server{
config: config,
}
space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
pm := policy.FromSpace(space)
if pm == nil {
return newError("Policy not found in space.")
Expand Down
2 changes: 1 addition & 1 deletion proxy/vmess/inbound/inbound.go
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ func New(ctx context.Context, config *Config) (*Handler, error) {
sessionHistory: encoding.NewSessionHistory(ctx),
}

space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
handler.inboundHandlerManager = proxyman.InboundHandlerManagerFromSpace(space)
if handler.inboundHandlerManager == nil {
return newError("InboundHandlerManager is not found is space.")
Expand Down
2 changes: 1 addition & 1 deletion proxy/vmess/outbound/outbound.go
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ func New(ctx context.Context, config *Config) (*Handler, error) {
serverPicker: protocol.NewRoundRobinServerPicker(serverList),
}

space.OnInitialize(func() error {
space.On(app.SpaceInitializing, func(interface{}) error {
pm := policy.FromSpace(space)
if pm == nil {
return newError("Policy is not found in space.")
Expand Down

0 comments on commit fd8db49

Please sign in to comment.