<commit_before>package net4g

import (
	"github.com/carsonsx/log4g"
	"reflect"
	"runtime/debug"
	"sync"
)

func Dispatch(dispatchers []*dispatcher, req NetReq, res NetRes) {
	for _, p := range dispatchers {
		if p.running {
			p.dispatchChan <- &dispatchData{req: req, res: res}
		}
	}
}

func AddHandler(dispatcher *dispatcher, v interface{}, h func(req NetReq, res NetRes)) {
	dispatcher.AddHandler(h, reflect.TypeOf(v))
}

func NewDispatcher(name string) *dispatcher {
	p := new(dispatcher)
	p.Name = name
	p.dispatchChan = make(chan *dispatchData, 1000)
	p.sessionClosedChan = make(chan NetSession, 100)
	p.destroyChan = make(chan bool, 1)
	p.typeHandlers = make(map[reflect.Type]func(req NetReq, res NetRes))
	p.run()
	log4g.Info("new a %s dispatcher", name)
	return p
}

type dispatchData struct {
	req NetReq
	res NetRes
}

type dispatcher struct {
	Name                    string
	serializer              Serializer
	mgr                     *NetManager
	globalHandlers          []func(req NetReq, res NetRes)
	typeHandlers            map[reflect.Type]func(req NetReq, res NetRes)
	before_interceptors     []func(req NetReq, res NetRes)
	after_interceptors      []func(req NetReq, res NetRes)
	dispatchChan            chan *dispatchData
	sessionClosedChan       chan NetSession
	connectionClosedHandler func(session NetSession)
	destroyChan             chan bool
	destroyHandler          func()
	running                 bool
	wg                      sync.WaitGroup
}

func (p *dispatcher) AddHandler(h func(req NetReq, res NetRes), t ...reflect.Type) {
	if len(t) > 0 {
		p.typeHandlers[t[0]] = h
		log4g.Info("added a handler for %v", t[0])
	} else {
		p.globalHandlers = append(p.globalHandlers, h)
		log4g.Info("added global handler")
	}
}

func (p *dispatcher) OnConnectionClosed(h func(session NetSession)) {
	p.connectionClosedHandler = h
}

func (p *dispatcher) OnDestroy(h func()) {
	p.destroyHandler = h
}

func (p *dispatcher) dispatch(msg *dispatchData) {

	// safe the user handler to avoid the whole server down
	defer func() {
		if r := recover(); r != nil {
			log4g.Error("********************* Handler Panic *********************")
			log4g.Error(r)
			debug.PrintStack()
			msg.res.Close()
			log4g.Error("********************* Handler Panic *********************")
		}
	}()

	for _, i := range p.before_interceptors {
		i(msg.req, msg.res)
	}

	for _, h := range p.globalHandlers {
		h(msg.req, msg.res)
	}

	t := reflect.TypeOf(msg.req.Msg())
	if h, ok := p.typeHandlers[t]; ok {
		log4g.Trace("dispatcher[%s] is dispatching %v to handler ", p.Name, t)
		h(msg.req, msg.res)
	}

	for _, i := range p.after_interceptors {
		i(msg.req, msg.res)
	}
}

func (p *dispatcher) run() {
	p.wg.Add(1)
	// one dispatcher, one goroutine
	go func() {
		defer p.wg.Done()<commit_msg>update log<commit_after>package net4g

import (
	"github.com/carsonsx/log4g"
	"reflect"
	"runtime/debug"
	"sync"
)

func Dispatch(dispatchers []*dispatcher, req NetReq, res NetRes) {
	for _, p := range dispatchers {
		if p.running {
			p.dispatchChan <- &dispatchData{req: req, res: res}
		}
	}
}

func AddHandler(dispatcher *dispatcher, v interface{}, h func(req NetReq, res NetRes)) {
	dispatcher.AddHandler(h, reflect.TypeOf(v))
}

func NewDispatcher(name string) *dispatcher {
	p := new(dispatcher)
	p.Name = name
	p.dispatchChan = make(chan *dispatchData, 1000)
	p.sessionClosedChan = make(chan NetSession, 100)
	p.destroyChan = make(chan bool, 1)
	p.typeHandlers = make(map[reflect.Type]func(req NetReq, res NetRes))
	p.run()
	log4g.Info("new a %s dispatcher", name)
	return p
}

type dispatchData struct {
	req NetReq
	res NetRes
}

type dispatcher struct {
	Name                    string
	serializer              Serializer
	mgr                     *NetManager
	globalHandlers          []func(req NetReq, res NetRes)
	typeHandlers            map[reflect.Type]func(req NetReq, res NetRes)
	before_interceptors     []func(req NetReq, res NetRes)
	after_interceptors      []func(req NetReq, res NetRes)
	dispatchChan            chan *dispatchData
	sessionClosedChan       chan NetSession
	connectionClosedHandler func(session NetSession)
	destroyChan             chan bool
	destroyHandler          func()
	running                 bool
	wg                      sync.WaitGroup
}

func (p *dispatcher) AddHandler(h func(req NetReq, res NetRes), t ...reflect.Type) {
	if len(t) > 0 {
		p.typeHandlers[t[0]] = h
		log4g.Info("added a handler for %v", t[0])
	} else {
		p.globalHandlers = append(p.globalHandlers, h)
		log4g.Info("added global handler")
	}
}

func (p *dispatcher) OnConnectionClosed(h func(session NetSession)) {
	p.connectionClosedHandler = h
}

func (p *dispatcher) OnDestroy(h func()) {
	p.destroyHandler = h
}

func (p *dispatcher) dispatch(msg *dispatchData) {

	// safe the user handler to avoid the whole server down
	defer func() {
		if r := recover(); r != nil {
			log4g.Error("********************* Handler Panic *********************")
			log4g.Error(r)
			log4g.Error(string(debug.Stack()))
			msg.res.Close()
			log4g.Error("********************* Handler Panic *********************")
		}
	}()

	for _, i := range p.before_interceptors {
		i(msg.req, msg.res)
	}

	for _, h := range p.globalHandlers {
		h(msg.req, msg.res)
	}

	t := reflect.TypeOf(msg.req.Msg())
	if h, ok := p.typeHandlers[t]; ok {
		log4g.Trace("dispatcher[%s] is dispatching %v to handler ", p.Name, t)
		h(msg.req, msg.res)
	}

	for _, i := range p.after_interceptors {
		i(msg.req, msg.res)
	}
}

func (p *dispatcher) run() {
	p.wg.Add(1)
	// one dispatcher, one goroutine
	go func() {
		defer p.wg.Done()
{'fraction_non_alphanumeric': 0.1536128456735058, 'fraction_numerical': 0.007671721677074041, 'mean_word_length': 3.322282189668466, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3894152', 'n_tokens_mistral': 2087, 'n_tokens_neox': 1940, 'n_words': 492}