Merge pull request #8 from divebomb/master

Imp: add max buffer length 

getty.go

@@ -29,14 +29,16 @@ type Reader interface {
 	// Parse tcp/udp/websocket pkg from buffer and if possible return a complete pkg.
 	// When receiving a tcp network streaming segment, there are 4 cases as following:
 	// case 1: a error found in the streaming segment;
-	// case 2: can not unmarshal a pkg from the streaming segment;
-	// case 3: just unmarshal a pkg from the streaming segment;
-	// case 4: unmarshal more than one pkg from the streaming segment;
+	// case 2: can not unmarshal a pkg header from the streaming segment;
+	// case 3: unmarshal a pkg header but can not unmarshal a pkg from the streaming segment;
+	// case 4: just unmarshal a pkg from the streaming segment;
+	// case 5: unmarshal more than one pkg from the streaming segment;
 	//
 	// The return value is (nil, 0, error) as case 1.
 	// The return value is (nil, 0, nil) as case 2.
-	// The return value is (pkg, pkgLen, nil) as case 3.
-	// The handleTcpPackage may invoke func Read many times as case 4.
+	// The return value is (nil, pkgLen, nil) as case 3.
+	// The return value is (pkg, pkgLen, nil) as case 4.
+	// The handleTcpPackage may invoke func Read many times as case 5.
 	Read(Session, []byte) (interface{}, int, error)
 }
 

session.go

@@ -664,6 +664,7 @@ func (s *session) handleTCPPackage() error {
 			}
 			// pkg, err = s.pkgHandler.Read(s, pktBuf)
 			pkg, pkgLen, err = s.reader.Read(s, pktBuf.Bytes())
+			// for case 3/case 4
 			if err == nil && s.maxMsgLen > 0 && pkgLen > int(s.maxMsgLen) {
 				err = perrors.Errorf("pkgLen %d > session max message len %d", pkgLen, s.maxMsgLen)
 			}
@@ -674,15 +675,15 @@ func (s *session) handleTCPPackage() error {
 				exit = true
 				break
 			}
-			// handle case 2
+			// handle case 2/case 3
 			if pkg == nil {
 				break
 			}
-			// handle case 3
+			// handle case 4
 			s.UpdateActive()
 			s.addTask(pkg)
 			pktBuf.Next(pkgLen)
-			// continue to handle case 4
+			// continue to handle case 5
 		}
 		if exit {
 			break
@@ -695,23 +696,24 @@ func (s *session) handleTCPPackage() error {
 // get package from udp packet
 func (s *session) handleUDPPackage() error {
 	var (
-		ok       bool
-		err      error
-		netError net.Error
-		conn     *gettyUDPConn
-		bufLen   int
-		buf      []byte
-		addr     *net.UDPAddr
-		pkgLen   int
-		pkg      interface{}
+		ok        bool
+		err       error
+		netError  net.Error
+		conn      *gettyUDPConn
+		bufLen    int
+		maxBufLen int
+		buf       []byte
+		addr      *net.UDPAddr
+		pkgLen    int
+		pkg       interface{}
 	)
 
 	conn = s.Connection.(*gettyUDPConn)
-	bufLen = int(s.maxMsgLen + maxReadBufLen)
+	maxBufLen = int(s.maxMsgLen + maxReadBufLen)
 	if int(s.maxMsgLen<<1) < bufLen {
-		bufLen = int(s.maxMsgLen << 1)
+		maxBufLen = int(s.maxMsgLen << 1)
 	}
-	buf = make([]byte, bufLen)
+	buf = make([]byte, maxBufLen)
 	for {
 		if s.IsClosed() {
 			break

task_pool.go

@@ -46,6 +46,7 @@ func NewTaskPool(opts ...TaskPoolOption) *TaskPool {
 	for i := 0; i < p.tQNumber; i++ {
 		p.qArray[i] = make(chan task, p.tQLen)
 	}
+	p.start()
 
 	return p
 }