Merge pull request #20 from fangyincheng/del-rQ

Mod: don't use read queue

getty.go

@@ -154,7 +154,10 @@ type Session interface {
 	SetReader(Reader)
 	SetWriter(Writer)
 	SetCronPeriod(int)
+
+	// Deprecated: don't use read queue.
 	SetRQLen(int)
+
 	SetWQLen(int)
 	SetWaitTime(time.Duration)
 	SetTaskPool(*gxsync.TaskPool)

session.go

@@ -66,8 +66,7 @@ type session struct {
 	reader Reader // @reader should be nil when @conn is a gettyWSConn object.
 	writer Writer
 
-	// read & write
-	rQ chan interface{}
+	// write
 	wQ chan interface{}
 
 	// handle logic
@@ -275,17 +274,8 @@ func (s *session) SetCronPeriod(period int) {
 	s.period = time.Duration(period) * time.Millisecond
 }
 
-// set @session's read queue size
-func (s *session) SetRQLen(readQLen int) {
-	if readQLen < 1 {
-		panic("@readQLen < 1")
-	}
-
-	s.lock.Lock()
-	defer s.lock.Unlock()
-	s.rQ = make(chan interface{}, readQLen)
-	log.Debug("%s, [session.SetRQLen] rQ{len:%d, cap:%d}", s.Stat(), len(s.rQ), cap(s.rQ))
-}
+// Deprecated: don't use read queue.
+func (s *session) SetRQLen(readQLen int) {}
 
 // set @session's Write queue size
 func (s *session) SetWQLen(writeQLen int) {
@@ -466,10 +456,6 @@ func (s *session) run() {
 		s.wQ = make(chan interface{}, defaultQLen)
 	}
 
-	if s.rQ == nil && s.tPool == nil {
-		s.rQ = make(chan interface{}, defaultQLen)
-	}
-
 	// call session opened
 	s.UpdateActive()
 	if err := s.listener.OnOpen(s); err != nil {
@@ -492,7 +478,6 @@ func (s *session) handleLoop() {
 		wsConn *gettyWSConn
 		// start  time.Time
 		counter gxtime.CountWatch
-		inPkg   interface{}
 		outPkg  interface{}
 	)
 
@@ -522,8 +507,8 @@ LOOP:
 		case <-s.done:
 			// this case branch assure the (session)handleLoop gr will exit before (session)handlePackage gr.
 			if atomic.LoadInt32(&(s.grNum)) == 1 { // make sure @(session)handlePackage goroutine has been closed.
-				if len(s.rQ) == 0 && len(s.wQ) == 0 {
-					log.Infof("%s, [session.handleLoop] got done signal. Both rQ and wQ are nil.", s.Stat())
+				if len(s.wQ) == 0 {
+					log.Infof("%s, [session.handleLoop] got done signal. wQ is nil.", s.Stat())
 					break LOOP
 				}
 				counter.Start()
@@ -534,18 +519,6 @@ LOOP:
 				}
 			}
 
-		case inPkg = <-s.rQ:
-			// read the s.rQ and assure (session)handlePackage gr will not block by (session)rQ.
-			if flag {
-				log.Debugf("%#v <-s.rQ", inPkg)
-				pkg := inPkg
-				// go s.listener.OnMessage(s, pkg)
-				s.listener.OnMessage(s, pkg)
-				s.incReadPkgNum()
-			} else {
-				log.Infof("[session.handleLoop] drop readin package{%#v}", inPkg)
-			}
-
 		case outPkg = <-s.wQ:
 			if flag {
 				if err = s.writer.Write(s, outPkg); err != nil {
@@ -573,14 +546,15 @@ LOOP:
 }
 
 func (s *session) addTask(pkg interface{}) {
+	f := func() {
+		s.listener.OnMessage(s, pkg)
+		s.incReadPkgNum()
+	}
 	if s.tPool != nil {
-		s.tPool.AddTask(func() {
-			s.listener.OnMessage(s, pkg)
-			s.incReadPkgNum()
-		})
-	} else {
-		s.rQ <- pkg
+		s.tPool.AddTask(f)
+		return
 	}
+	f()
 }
 
 func (s *session) handlePackage() {
@@ -853,10 +827,6 @@ func (s *session) gc() {
 			close(s.wQ)
 			s.wQ = nil
 		}
-		if s.rQ != nil {
-			close(s.rQ)
-			s.rQ = nil
-		}
 		s.Connection.close((int)((int64)(s.wait)))
 	}
 	s.lock.Unlock()