mirror of https://github.com/qwqdanchun/nps.git
fix buffer size bug
This commit is contained in:
parent
53c2e472ae
commit
41c282b38b
|
@ -27,10 +27,13 @@ func (Self *BasePackager) NewPac(contents ...interface{}) (err error) {
|
|||
case nil:
|
||||
Self.Content = Self.Content[:0]
|
||||
case []byte:
|
||||
Self.Content = append(Self.Content, content.([]byte)...)
|
||||
err = Self.appendByte(content.([]byte))
|
||||
case string:
|
||||
Self.Content = append(Self.Content, []byte(content.(string))...)
|
||||
Self.Content = append(Self.Content, []byte(CONN_DATA_SEQ)...)
|
||||
err = Self.appendByte([]byte(content.(string)))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = Self.appendByte([]byte(CONN_DATA_SEQ))
|
||||
default:
|
||||
err = Self.marshal(content)
|
||||
}
|
||||
|
@ -39,6 +42,18 @@ func (Self *BasePackager) NewPac(contents ...interface{}) (err error) {
|
|||
return
|
||||
}
|
||||
|
||||
func (Self *BasePackager) appendByte(data []byte) (err error) {
|
||||
m := len(Self.Content)
|
||||
n := m + len(data)
|
||||
if n <= cap(Self.Content) {
|
||||
Self.Content = Self.Content[0:n] // grow the length for copy
|
||||
copy(Self.Content[m:n], data)
|
||||
return nil
|
||||
} else {
|
||||
return bytes.ErrTooLarge
|
||||
}
|
||||
}
|
||||
|
||||
//似乎这里涉及到父类作用域问题,当子类调用父类的方法时,其struct仅仅为父类的
|
||||
func (Self *BasePackager) Pack(writer io.Writer) (err error) {
|
||||
err = binary.Write(writer, binary.LittleEndian, Self.Length)
|
||||
|
@ -53,12 +68,12 @@ func (Self *BasePackager) Pack(writer io.Writer) (err error) {
|
|||
//Unpack 会导致传入的数字类型转化成float64!!
|
||||
//主要原因是json unmarshal并未传入正确的数据类型
|
||||
func (Self *BasePackager) UnPack(reader io.Reader) (err error) {
|
||||
Self.Content = pool.CopyBuff.Get()
|
||||
Self.clean()
|
||||
err = binary.Read(reader, binary.LittleEndian, &Self.Length)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
Self.Content = pool.CopyBuff.Get()
|
||||
Self.Content = Self.Content[:Self.Length]
|
||||
//n, err := io.ReadFull(reader, Self.Content)
|
||||
//if n != int(Self.Length) {
|
||||
|
@ -73,7 +88,7 @@ func (Self *BasePackager) marshal(content interface{}) (err error) {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
Self.Content = append(Self.Content, tmp...)
|
||||
err = Self.appendByte(tmp)
|
||||
return
|
||||
}
|
||||
|
||||
|
@ -92,7 +107,7 @@ func (Self *BasePackager) setLength() {
|
|||
|
||||
func (Self *BasePackager) clean() {
|
||||
Self.Length = 0
|
||||
Self.Content = Self.Content[:0]
|
||||
Self.Content = Self.Content[:0] // reset length
|
||||
}
|
||||
|
||||
func (Self *BasePackager) Split() (strList []string) {
|
||||
|
@ -162,7 +177,6 @@ func (Self *MuxPackager) Pack(writer io.Writer) (err error) {
|
|||
}
|
||||
|
||||
func (Self *MuxPackager) UnPack(reader io.Reader) (err error) {
|
||||
Self.Length = 0
|
||||
err = binary.Read(reader, binary.LittleEndian, &Self.Flag)
|
||||
if err != nil {
|
||||
return
|
||||
|
|
|
@ -264,13 +264,19 @@ func GetPortByAddr(addr string) int {
|
|||
return p
|
||||
}
|
||||
|
||||
func CopyBuffer(dst io.Writer, src io.Reader,connId int32) (written int64, err error) {
|
||||
buf := pool.GetBufPoolCopy()
|
||||
defer pool.PutBufPoolCopy(buf)
|
||||
func CopyBuffer(dst io.Writer, src io.Reader, connId int32) (written int64, err error) {
|
||||
buf := pool.CopyBuff.Get()
|
||||
defer pool.CopyBuff.Put(buf)
|
||||
for {
|
||||
nr, er := src.Read(buf)
|
||||
if er != nil {
|
||||
if er != io.EOF {
|
||||
err = er
|
||||
}
|
||||
break
|
||||
}
|
||||
if nr > 0 {
|
||||
logs.Warn("write",connId, nr, string(buf[0:10]))
|
||||
logs.Warn("write", connId, nr, string(buf[0:10]))
|
||||
nw, ew := dst.Write(buf[0:nr])
|
||||
if nw > 0 {
|
||||
written += int64(nw)
|
||||
|
@ -284,12 +290,6 @@ func CopyBuffer(dst io.Writer, src io.Reader,connId int32) (written int64, err e
|
|||
break
|
||||
}
|
||||
}
|
||||
if er != nil {
|
||||
if er != io.EOF {
|
||||
err = er
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
return written, err
|
||||
}
|
||||
|
|
|
@ -228,7 +228,7 @@ func (s *Mux) Close() error {
|
|||
select {
|
||||
case s.closeChan <- struct{}{}:
|
||||
}
|
||||
s.closeChan <- struct{}{}
|
||||
//s.closeChan <- struct{}{}
|
||||
close(s.writeQueue)
|
||||
close(s.newConnCh)
|
||||
return s.conn.Close()
|
||||
|
|
|
@ -66,13 +66,14 @@ type CopyBufferPool struct {
|
|||
func (Self *CopyBufferPool) New() {
|
||||
Self.pool = sync.Pool{
|
||||
New: func() interface{} {
|
||||
return make([]byte, PoolSizeCopy)
|
||||
return make([]byte, PoolSizeCopy, PoolSizeCopy)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (Self *CopyBufferPool) Get() []byte {
|
||||
return Self.pool.Get().([]byte)
|
||||
buf := Self.pool.Get().([]byte)
|
||||
return buf[:cap(buf)] // grow to capacity
|
||||
}
|
||||
|
||||
func (Self *CopyBufferPool) Put(x []byte) {
|
||||
|
|
Loading…
Reference in New Issue