nnet-old/nnet/connection.go

package nnet

import (
	"errors"
	"git.noahlan.cn/northlan/nnet/log"
	"git.noahlan.cn/northlan/nnet/nface"
	"git.noahlan.cn/northlan/nnet/packet"
	"git.noahlan.cn/northlan/nnet/pipeline"
	"git.noahlan.cn/northlan/nnet/session"
	"net"
	"sync/atomic"
	"time"
)

var (
	_ nface.IConnection = (*Connection)(nil)

	ErrCloseClosedSession = errors.New("close closed session")
	// ErrBrokenPipe represents the low-level connection has broken.
	ErrBrokenPipe = errors.New("broken low-level pipe")
	// ErrBufferExceed indicates that the current session buffer is full and
	// can not receive more data.
	ErrBufferExceed = errors.New("session send buffer exceed")
)

type (
	Connection struct {
		session nface.ISession // Session
		server  *Server        // Server 引用

		conn            net.Conn // low-level conn fd
		status          int32    // 连接状态
		lastMid         uint64   // 最近一次消息ID
		lastHeartbeatAt int64    // 最近一次心跳时间

		chDie  chan struct{}       // 停止通道
		chSend chan pendingMessage // 消息发送通道

		pipeline pipeline.Pipeline // 消息管道
	}

	pendingMessage struct {
		typ     interface{} // message type
		route   string      // message route
		mid     uint64      // response message id
		payload interface{} // payload
	}
)

func newConnection(server *Server, conn net.Conn, pipeline pipeline.Pipeline) *Connection {
	r := &Connection{
		conn:   conn,
		server: server,
		status: nface.StatusStart,

		lastHeartbeatAt: time.Now().Unix(),

		chDie:  make(chan struct{}),
		chSend: make(chan pendingMessage, 512),

		pipeline: pipeline,
	}

	// binding session
	r.session = session.New()
	return r
}

func (r *Connection) send(m pendingMessage) (err error) {
	defer func() {
		if e := recover(); e != nil {
			err = ErrBrokenPipe
		}
	}()
	r.chSend <- m
	return err
}

func (r *Connection) Server() nface.IServer {
	return r.server
}

func (r *Connection) Status() int32 {
	return atomic.LoadInt32(&r.status)
}

func (r *Connection) SetStatus(s int32) {
	atomic.StoreInt32(&r.status, s)
}

func (r *Connection) Conn() net.Conn {
	return r.conn
}

func (r *Connection) ID() int64 {
	return r.session.ID()
}

func (r *Connection) setLastHeartbeatAt(t int64) {
	atomic.StoreInt64(&r.lastHeartbeatAt, t)
}

func (r *Connection) Session() nface.ISession {
	return r.session
}

func (r *Connection) write() {
	ticker := time.NewTicker(r.server.HeartbeatInterval)

	chWrite := make(chan []byte, 1024)

	defer func() {
		ticker.Stop()
		close(r.chSend)
		close(chWrite)
		_ = r.Close()

		log.Debugf("Connection write goroutine exit, ConnID=%d, SessionUID=%d", r.ID(), r.session.UID())
	}()

	for {
		select {
		case <-ticker.C:
			// TODO heartbeat enable control
			deadline := time.Now().Add(-2 * r.server.HeartbeatInterval).Unix()
			if atomic.LoadInt64(&r.lastHeartbeatAt) < deadline {
				log.Debugf("Session heartbeat timeout, LastTime=%d, Deadline=%d", atomic.LoadInt64(&r.lastHeartbeatAt), deadline)
				return
			}
			// TODO heartbeat data
			chWrite <- []byte{}
		case data := <-r.chSend:
			// marshal packet body (data)
			payload, err := r.server.Serializer.Marshal(data.payload)
			if err != nil {
				log.Errorf("message body marshal err: %v", err)
				break
			}
			// TODO new message and pipeline

			if pipe := r.pipeline; pipe != nil {
				err := pipe.Outbound().Process(r)
				if err != nil {
					log.Errorf("broken pipeline err: %s", err.Error())
					break
				}
			}

			// TODO encode message ? message processor ?

			// packet pack data
			p, err := r.server.Packer.Pack(packet.Data, payload)
			if err != nil {
				log.Error(err.Error())
				break
			}
			chWrite <- p
		case data := <-chWrite:
			// 回写数据
			if _, err := r.conn.Write(data); err != nil {
				log.Error(err.Error())
				return
			}
		case <-r.chDie: // connection close signal
			return
		case <-r.server.DieChan: // application quit signal
			return
		}
	}
}

func (r *Connection) Close() error {
	if r.Status() == nface.StatusClosed {
		return ErrCloseClosedSession
	}
	r.SetStatus(nface.StatusClosed)

	log.Debugf("close connection, ID: %d", r.ID())

	select {
	case <-r.chDie:
	default:
		close(r.chDie)
		// TODO lifetime
	}
	return r.conn.Close()
}