<commit_before>/**
 * hub, core of chat application
 */

package tok

import (
	"expvar"
	"log"
)

var (
	expOnline = expvar.NewInt("tokOnline")
	expUp     = expvar.NewInt("tokUp")
	expDown   = expvar.NewInt("tokDown")
	expEnq    = expvar.NewInt("tokEnq")
	expDeq    = expvar.NewInt("tokDeq")
)

type fatFrame struct {
	frame *frame //frame to be sent
	ttl   uint32
	chErr chan error //channel to read send result from
}

type frame struct {
	uid  interface{}
	data []byte
}

//Config to create new Hub
type HubConfig struct {
	Actor Actor //Actor implement dispatch logic
	Q     Queue //Message Queue, if nil, message to offline user will not be cached
	Sso   bool  //If it's true, new connection  with same uid will kick off old ones
}

//Dispatch message between connections
type Hub struct {
	sso           bool
	actor         Actor
	q             Queue
	cons          map[interface{}][]*connection //connection list
	chUp          chan *frame
	chDown        chan *fatFrame //for online user
	chDown2       chan *fatFrame //for all user
	chConState    chan *conState
	chReadSignal  chan interface{}
	chQueryOnline chan chan []interface{}
}

func createHub(actor Actor, q Queue, sso bool) *Hub {
	if READ_TIMEOUT > 0 {
		log.Println("[tok] read timeout is enabled, make sure it's greater than your client ping interval. otherwise you'll get read timeout err")
	} else {
		if actor.Ping() == nil {
			log.Fatalln("[tok] both read timeout and server ping have been disabled, server socket resource leak might happen")
		}
	}

	hub := &Hub{
		sso:           sso,
		actor:         actor,
		q:             q,
		cons:          make(map[interface{}][]*connection),
		chUp:          make(chan *frame),
		chDown:        make(chan *fatFrame),
		chDown2:       make(chan *fatFrame),
		chConState:    make(chan *conState),
		chReadSignal:  make(chan interface{}),
		chQueryOnline: make(chan chan []interface{}),
	}
	go hub.run()
	return hub
}

func (p *Hub) run() {
	for {
		select {
		case state := <-p.chConState:
			//			log.Printf("connection state change: %v, %v \n", state.online, &state.con)

			if state.online {
				p.goOnline(state.con)
			} else {
				p.goOffline(state.con)
			}
			count := int64(len(p.cons))
			expOnline.Set(count)
		case f := <-p.chUp:
			//			log.Println("up data")
			expUp.Add(1)
			go p.actor.OnReceive(f.uid, f.data)
		case ff := <-p.chDown:
			if len(p.cons[ff.frame.uid]) > 0 {
				expDown.Add(1)
				go p.down(ff)
			} else {
				ff.chErr <- ErrOffline
				close(ff.chErr)
			}
		case ff := <-p.chDown2:
			if len(p.cons[ff.frame.uid]) > 0 {
				expDown.Add(1)
				go p.down(ff)
			} else {
				expEnq.Add(1)
				go p.cache(ff)
			}

		case uid := <-p.chReadSignal:
			//only pop msg for online user
			if len(p.cons[uid]) > 0 {
				go p.popMsg(uid)
			}
		case chOnline := <-p.chQueryOnline:
			result := make([]interface{}, 0, len(p.cons))
			for uid := range p.cons {
				result = append(result, uid)
			}
			chOnline <- result
			close(chOnline)
		}
	}
}

func (p *Hub) popMsg(uid interface{}) {
	if p.q == nil {
		return
	}

	for {
		b, err := p.q.Deq(uid)
		if err != nil {
			log.Println("deq error", err)
			return
		}
		if b == nil {
			//no more data in queue
			return
		}
		expDeq.Add(1)
		if err := p.Send(uid, b); err != nil {
			log.Println("send err after deq")
			return
		}
	}
}

//Send message to someone,
//ttl is expiry seconds. 0 means forever
//if ttl >= 0 and user is offline, message will be cached for ttl seconds
//if ttl < 0 and user is offline, ErrOffline will be returned
//if ttl >=0 and user is online, but error occurred during send, message will be cached
//if ttl < 0 and user is online, but error occurred during send, the error will be returned
func (p *Hub) Send(to interface{}, b []byte, ttl ...int) error {
	t := 0
	if len(ttl) > 0 {
		t = ttl[0]
	}

	ff := &fatFrame{frame: &frame{uid: to, data: b}, ttl: uint32(t), chErr: make(chan error)}
	if t > 0 {
		p.chDown2 <- ff<commit_msg>minor update<commit_after>/**
 * hub, core of chat application
 */

package tok

import (
	"expvar"
	"log"
)

var (
	expOnline = expvar.NewInt("tokOnline")
	expUp     = expvar.NewInt("tokUp")
	expDown   = expvar.NewInt("tokDown")
	expEnq    = expvar.NewInt("tokEnq")
	expDeq    = expvar.NewInt("tokDeq")
)

type fatFrame struct {
	frame *frame //frame to be sent
	ttl   uint32
	chErr chan error //channel to read send result from
}

type frame struct {
	uid  interface{}
	data []byte
}

//Config to create new Hub
type HubConfig struct {
	Actor Actor //Actor implement dispatch logic
	Q     Queue //Message Queue, if nil, message to offline user will not be cached
	Sso   bool  //If it's true, new connection  with same uid will kick off old ones
}

//Dispatch message between connections
type Hub struct {
	sso           bool
	actor         Actor
	q             Queue
	cons          map[interface{}][]*connection //connection list
	chUp          chan *frame
	chDown        chan *fatFrame //for online user
	chDown2       chan *fatFrame //for all user
	chConState    chan *conState
	chReadSignal  chan interface{}
	chQueryOnline chan chan []interface{}
}

func createHub(actor Actor, q Queue, sso bool) *Hub {
	if READ_TIMEOUT > 0 {
		log.Println("[tok] read timeout is enabled, make sure it's greater than your client ping interval. otherwise you'll get read timeout err")
	} else {
		if actor.Ping() == nil {
			log.Fatalln("[tok] both read timeout and server ping have been disabled, server socket resource leak might happen")
		}
	}

	hub := &Hub{
		sso:           sso,
		actor:         actor,
		q:             q,
		cons:          make(map[interface{}][]*connection),
		chUp:          make(chan *frame),
		chDown:        make(chan *fatFrame),
		chDown2:       make(chan *fatFrame),
		chConState:    make(chan *conState),
		chReadSignal:  make(chan interface{}),
		chQueryOnline: make(chan chan []interface{}),
	}
	go hub.run()
	return hub
}

func (p *Hub) run() {
	for {
		select {
		case state := <-p.chConState:
			//			log.Printf("connection state change: %v, %v \n", state.online, &state.con)

			if state.online {
				p.goOnline(state.con)
			} else {
				p.goOffline(state.con)
			}
			count := int64(len(p.cons))
			expOnline.Set(count)
		case f := <-p.chUp:
			//			log.Println("up data")
			expUp.Add(1)
			go p.actor.OnReceive(f.uid, f.data)
		case ff := <-p.chDown:
			if len(p.cons[ff.frame.uid]) > 0 {
				expDown.Add(1)
				go p.down(ff)
			} else {
				ff.chErr <- ErrOffline
				close(ff.chErr)
			}
		case ff := <-p.chDown2:
			if len(p.cons[ff.frame.uid]) > 0 {
				expDown.Add(1)
				go p.down(ff)
			} else {
				expEnq.Add(1)
				go p.cache(ff)
			}

		case uid := <-p.chReadSignal:
			//only pop msg for online user
			if len(p.cons[uid]) > 0 {
				go p.popMsg(uid)
			}
		case chOnline := <-p.chQueryOnline:
			result := make([]interface{}, 0, len(p.cons))
			for uid := range p.cons {
				result = append(result, uid)
			}
			chOnline <- result
			close(chOnline)
		}
	}
}

func (p *Hub) popMsg(uid interface{}) {
	if p.q == nil {
		return
	}

	for {
		b, err := p.q.Deq(uid)
		if err != nil {
			log.Println("deq error", err)
			return
		}
		if b == nil {
			//no more data in queue
			return
		}
		expDeq.Add(1)
		if err := p.Send(uid, b); err != nil {
			log.Println("send err after deq")
			return
		}
	}
}

//Send message to someone,
//ttl is expiry seconds. 0 means forever
//if ttl >= 0 and user is offline, message will be cached for ttl seconds
//if ttl < 0 and user is offline, ErrOffline will be returned
//if ttl >=0 and user is online, but error occurred during send, message will be cached
//if ttl < 0 and user is online, but error occurred during send, the error will be returned
func (p *Hub) Send(to interface{}, b []byte, ttl ...int) error {
	t := -1
	if len(ttl) > 0 {
		t = ttl[0]
	}

	ff := &fatFrame{frame: &frame{uid: to, data: b}, ttl: uint32(t), chErr: make(chan error)}
	if t > 0 {
		p.chDown2 <- ff
{'fraction_non_alphanumeric': 0.13265048052604958, 'fraction_numerical': 0.007334344967121902, 'mean_word_length': 2.63632183908046, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 19, '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': '2685668', 'n_tokens_mistral': 3225, 'n_tokens_neox': 2830, 'n_words': 904}