diff --git a/session.go b/session.go index 7528352..3feb717 100644 --- a/session.go +++ b/session.go @@ -31,11 +31,12 @@ type Session struct { tbf chan struct{} // tokenbuffer frameQueues map[uint32][]Frame // stream input frame queue - die chan struct{} // flag session has died - chAccepts chan *Stream - chActiveClose chan uint32 - dataReady int32 // flag data has arrived - mu sync.Mutex + die chan struct{} // flag session has died + chAccepts chan *Stream + chClosedStream chan uint32 + + dataReady int32 // flag data has arrived + mu sync.Mutex } func newSession(config *Config, conn io.ReadWriteCloser, client bool) *Session { @@ -47,7 +48,7 @@ func newSession(config *Config, conn io.ReadWriteCloser, client bool) *Session { s.frameQueues = make(map[uint32][]Frame) s.rdEvents = make(map[uint32]chan struct{}) s.chAccepts = make(chan *Stream, defaultAcceptBacklog) - s.chActiveClose = make(chan uint32, defaultCloseWait) + s.chClosedStream = make(chan uint32, defaultCloseWait) s.tbf = make(chan struct{}, config.MaxFrameTokens) for i := 0; i < config.MaxFrameTokens; i++ { s.tbf <- struct{}{} @@ -131,7 +132,10 @@ func (s *Session) NumStreams() int { // notify the session that a session has closed func (s *Session) streamClosed(sid uint32) { - s.chActiveClose <- sid + select { + case s.chClosedStream <- sid: + case <-s.die: + } } // nonblocking read from session pool, for streams @@ -175,7 +179,7 @@ func (s *Session) readFrame(buffer []byte) (f Frame, err error) { func (s *Session) monitor() { for { select { - case sid := <-s.chActiveClose: + case sid := <-s.chClosedStream: s.mu.Lock() delete(s.streams, sid) delete(s.rdEvents, sid)