session.go

package link

import (
	"container/list"
	"errors"
	"net"
	"sync"
	"sync/atomic"
)

var ErrClosed = errors.New("link.Session closed")

type Session struct {
	id               uint64
	conn             net.Conn
	encoder          Encoder
	decoder          Decoder
	closeChan        chan int
	closeFlag        int32
	closeEventMutex  sync.Mutex
	closeCallbacks   *list.List
	State            interface{}
	SendPackageList  *list.List
	sendPackageMutex sync.Mutex
}

var globalSessionId uint64

func NewSession(conn net.Conn, codecType CodecType) *Session {
	session := &Session{
		id:              atomic.AddUint64(&globalSessionId, 1),
		conn:            conn,
		encoder:         codecType.NewEncoder(conn),
		decoder:         codecType.NewDecoder(conn),
		closeCallbacks:  list.New(),
		SendPackageList: list.New(),
	}
	return session
}

func (session *Session) Id() uint64     { return session.id }
func (session *Session) Conn() net.Conn { return session.conn }
func (session *Session) IsClosed() bool { return atomic.LoadInt32(&session.closeFlag) == 1 }

func (session *Session) Close() {
	if atomic.CompareAndSwapInt32(&session.closeFlag, 0, 1) {
		session.invokeCloseCallbacks()
		if session.closeChan != nil {
			close(session.closeChan)
		}
		session.conn.Close()
		if d, ok := session.encoder.(Disposeable); ok {
			d.Dispose()
		}
		if d, ok := session.decoder.(Disposeable); ok {
			d.Dispose()
		}
	}
}

func (session *Session) AppendSendPackage(msg interface{}) {
	session.sendPackageMutex.Lock()
	defer session.sendPackageMutex.Unlock()
	session.SendPackageList.PushBack(msg)
}

func (session *Session) ClearSendPackage() bool {
	if session.SendPackageList.Len() <= 0 {
		return false
	}
	session.sendPackageMutex.Lock()
	defer session.sendPackageMutex.Unlock()
	for i := session.SendPackageList.Front(); i != nil; i = i.Next() {
		session.Send(i.Value)
	}
	session.SendPackageList = list.New()
	return true
}

func (session *Session) Receive(msg interface{}) (err error) {
	if session.IsClosed() {
		return ErrClosed
	}
	err = session.decoder.Decode(msg)
	if err != nil {
		session.Close()
	}
	return
}

func (session *Session) Send(msg interface{}) (err error) {
	if session.IsClosed() {
		return ErrClosed
	}
	err = session.encoder.Encode(msg)
	if err != nil {
		session.Close()
	}
	return
}

type closeCallback struct {
	Handler interface{}
	Func    func()
}

func (session *Session) AddCloseCallback(handler interface{}, callback func()) {
	if session.IsClosed() {
		return
	}

	session.closeEventMutex.Lock()
	defer session.closeEventMutex.Unlock()

	session.closeCallbacks.PushBack(closeCallback{handler, callback})
}

func (session *Session) RemoveCloseCallback(handler interface{}) {
	if session.IsClosed() {
		return
	}

	session.closeEventMutex.Lock()
	defer session.closeEventMutex.Unlock()

	for i := session.closeCallbacks.Front(); i != nil; i = i.Next() {
		if i.Value.(closeCallback).Handler == handler {
			session.closeCallbacks.Remove(i)
			return
		}
	}
}

func (session *Session) invokeCloseCallbacks() {
	session.closeEventMutex.Lock()
	defer session.closeEventMutex.Unlock()

	for i := session.closeCallbacks.Front(); i != nil; i = i.Next() {
		callback := i.Value.(closeCallback)
		callback.Func()
	}
	session.ClearSendPackage()
}