600 lines
14 KiB
Go
600 lines
14 KiB
Go
// Copyright 2015 Tony Bai.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package cmpp
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net"
|
|
"os"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// errors for cmpp server
|
|
var (
|
|
ErrEmptyServerAddr = errors.New("cmpp server listen: empty server addr")
|
|
ErrNoHandlers = errors.New("cmpp server: no connection handler")
|
|
ErrUnsupportedPkt = errors.New("cmpp server read packet: receive a unsupported pkt")
|
|
)
|
|
|
|
type Packet struct {
|
|
Packer
|
|
*Conn
|
|
}
|
|
|
|
type Response struct {
|
|
*Packet
|
|
Packer
|
|
SeqId uint32
|
|
AfterSend func(error)
|
|
}
|
|
|
|
type Handler interface {
|
|
ServeCmpp(*Response, *Packet, *log.Logger) (bool, error)
|
|
}
|
|
|
|
// The HandlerFunc type is an adapter to allow the use of
|
|
// ordinary functions as Cmpp handlers. If f is a function
|
|
// with the appropriate signature, HandlerFunc(f) is a
|
|
// Handler object that calls f.
|
|
//
|
|
// The first return value indicates whether to invoke next handler in
|
|
// the chain of handlers.
|
|
//
|
|
// The second return value shows the error returned from the handler. And
|
|
// if it is non-nil, server will close the client connection
|
|
// after sending back the corresponding response.
|
|
type HandlerFunc func(*Response, *Packet, *log.Logger) (bool, error)
|
|
|
|
// ServeCmpp calls f(r, p).
|
|
func (f HandlerFunc) ServeCmpp(r *Response, p *Packet, l *log.Logger) (bool, error) {
|
|
return f(r, p, l)
|
|
}
|
|
|
|
type Server struct {
|
|
Addr string
|
|
Handler Handler
|
|
|
|
// protocol info
|
|
Typ Type
|
|
T time.Duration // interval betwwen two active tests
|
|
N int32 // continuous send times when no response back
|
|
|
|
// ErrorLog specifies an optional logger for errors accepting
|
|
// connections and unexpected behavior from handlers.
|
|
// If nil, logging goes to os.Stderr via the log package's
|
|
// standard logger.
|
|
ErrorLog *log.Logger
|
|
OnClose func(*Conn)
|
|
// SubmitWindow resolves the authenticated client's allowed in-flight Submit
|
|
// count. A nil resolver, or a value below two, preserves serial handling.
|
|
SubmitWindow func(*Conn) int
|
|
}
|
|
|
|
// A conn represents the server side of a Cmpp connection.
|
|
type conn struct {
|
|
*Conn
|
|
server *Server // the Server on which the connection arrived
|
|
|
|
// for active test
|
|
t time.Duration // interval betwwen two active tests
|
|
n int32 // continuous send times when no response back
|
|
done chan struct{}
|
|
exceed chan struct{}
|
|
counter int32
|
|
}
|
|
|
|
// Serve accepts incoming connections on the Listener l, creating a
|
|
// new service goroutine for each. The service goroutines read requests and
|
|
// then call srv.Handler to reply to them.
|
|
func (srv *Server) Serve(l net.Listener) error {
|
|
defer l.Close()
|
|
var tempDelay time.Duration // how long to sleep on accept failure
|
|
for {
|
|
rw, e := l.Accept()
|
|
if e != nil {
|
|
if ne, ok := e.(net.Error); ok && ne.Temporary() {
|
|
if tempDelay == 0 {
|
|
tempDelay = 5 * time.Millisecond
|
|
} else {
|
|
tempDelay *= 2
|
|
}
|
|
if max := 1 * time.Second; tempDelay > max {
|
|
tempDelay = max
|
|
}
|
|
srv.ErrorLog.Printf("accept error: %v; retrying in %v", e, tempDelay)
|
|
time.Sleep(tempDelay)
|
|
continue
|
|
}
|
|
return e
|
|
}
|
|
tempDelay = 0
|
|
c, err := srv.newConn(rw)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
|
|
srv.ErrorLog.Printf("accept a connection from %v\n", c.Conn.RemoteAddr())
|
|
go c.serve()
|
|
}
|
|
}
|
|
|
|
func (c *conn) readPacket() (*Response, error) {
|
|
readTimeout := time.Second * 2
|
|
i, err := c.Conn.RecvAndUnpackPkt(readTimeout)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
typ := c.Conn.Typ
|
|
|
|
var pkt *Packet
|
|
var rsp *Response
|
|
switch p := i.(type) {
|
|
case *CmppConnReqPkt:
|
|
if p.Version == V20 || p.Version == V21 || p.Version == V30 {
|
|
c.Conn.Typ = p.Version
|
|
typ = p.Version
|
|
}
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
if typ == V30 {
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &Cmpp3ConnRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp30 connect request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
} else {
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &Cmpp2ConnRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp20 connect request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
}
|
|
|
|
case *Cmpp2SubmitReqPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &Cmpp2SubmitRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp20 submit request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *Cmpp3SubmitReqPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &Cmpp3SubmitRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp30 submit request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *Cmpp2FwdReqPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &Cmpp2FwdRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp20 forward request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *Cmpp3FwdReqPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &Cmpp3FwdRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp30 forward request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *Cmpp2DeliverRspPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp20 deliver response from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *Cmpp3DeliverRspPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp30 deliver response from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *CmppActiveTestReqPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &CmppActiveTestRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp active request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *CmppActiveTestRspPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp active response from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *CmppTerminateReqPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
Packer: &CmppTerminateRspPkt{
|
|
SeqId: p.SeqId,
|
|
},
|
|
SeqId: p.SeqId,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp terminate request from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
|
|
case *CmppTerminateRspPkt:
|
|
pkt = &Packet{
|
|
Packer: p,
|
|
Conn: c.Conn,
|
|
}
|
|
|
|
rsp = &Response{
|
|
Packet: pkt,
|
|
}
|
|
c.server.ErrorLog.Printf("receive a cmpp terminate response from %v[%d]\n",
|
|
c.Conn.RemoteAddr(), p.SeqId)
|
|
default:
|
|
return nil, NewOpError(ErrUnsupportedPkt,
|
|
fmt.Sprintf("readPacket: receive unsupported packet type: %#v", p))
|
|
}
|
|
return rsp, nil
|
|
}
|
|
|
|
// Close the connection.
|
|
func (c *conn) close() {
|
|
p := &CmppTerminateReqPkt{}
|
|
|
|
err := c.Conn.SendPkt(p, <-c.Conn.SeqId)
|
|
if err != nil {
|
|
c.server.ErrorLog.Printf("send cmpp terminate request packet to %v error: %v\n", c.Conn.RemoteAddr(), err)
|
|
}
|
|
|
|
close(c.done)
|
|
c.server.ErrorLog.Printf("close connection with %v!\n", c.Conn.RemoteAddr())
|
|
c.Conn.Close()
|
|
}
|
|
|
|
func (c *conn) finishPacket(r *Response) error {
|
|
if _, ok := r.Packet.Packer.(*CmppActiveTestRspPkt); ok {
|
|
atomic.AddInt32(&c.counter, -1)
|
|
return nil
|
|
}
|
|
|
|
if r.Packer == nil {
|
|
// For response packet received, it need not
|
|
// to send anything back.
|
|
return nil
|
|
}
|
|
|
|
return c.Conn.SendPkt(r.Packer, r.SeqId)
|
|
}
|
|
|
|
func startActiveTest(c *conn) {
|
|
exceed, done := make(chan struct{}), make(chan struct{})
|
|
c.done = done
|
|
c.exceed = exceed
|
|
|
|
go func() {
|
|
t := time.NewTicker(c.t)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-done:
|
|
// once conn close, the goroutine should exit
|
|
return
|
|
case <-t.C:
|
|
// check whether c.counter exceeds
|
|
if atomic.LoadInt32(&c.counter) >= c.n {
|
|
c.server.ErrorLog.Printf("no cmpp active test response returned from %v for %d times!",
|
|
c.Conn.RemoteAddr(), c.n)
|
|
exceed <- struct{}{}
|
|
break
|
|
}
|
|
// send a active test packet to peer, increase the active test counter
|
|
p := &CmppActiveTestReqPkt{}
|
|
err := c.Conn.SendPkt(p, <-c.Conn.SeqId)
|
|
if err != nil {
|
|
c.server.ErrorLog.Printf("send cmpp active test request to %v error: %v", c.Conn.RemoteAddr(), err)
|
|
} else {
|
|
atomic.AddInt32(&c.counter, 1)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Serve a new connection.
|
|
func (c *conn) serve() {
|
|
defer func() {
|
|
if err := recover(); err != nil {
|
|
c.server.ErrorLog.Printf("panic serving %v: %v\n", c.Conn.RemoteAddr(), err)
|
|
}
|
|
}()
|
|
|
|
var submitGroup sync.WaitGroup
|
|
var submitSlots chan struct{}
|
|
fatal := make(chan error, 1)
|
|
defer func() {
|
|
// Why wait: a handler may persist the message and register receipt routing
|
|
// after the peer disconnects. Session cleanup must run after every accepted
|
|
// in-flight request finishes, otherwise a late handler can recreate stale state.
|
|
submitGroup.Wait()
|
|
c.close()
|
|
if c.server.OnClose != nil {
|
|
c.server.OnClose(c.Conn)
|
|
}
|
|
}()
|
|
|
|
// start a goroutine for sending active test.
|
|
startActiveTest(c)
|
|
|
|
for {
|
|
select {
|
|
case <-c.exceed:
|
|
return // close the connection.
|
|
case <-fatal:
|
|
return
|
|
default:
|
|
}
|
|
|
|
r, err := c.readPacket()
|
|
if err != nil {
|
|
if e, ok := err.(net.Error); ok && e.Timeout() {
|
|
continue
|
|
}
|
|
if errors.Is(err, io.EOF) {
|
|
break
|
|
}
|
|
c.server.ErrorLog.Printf(
|
|
"read/unpack packet failed remote=%v protocol=%s err_type=%T err=%v",
|
|
c.Conn.RemoteAddr(), c.Conn.Typ, err, err,
|
|
)
|
|
break
|
|
}
|
|
|
|
if isSubmitPacket(r.Packet.Packer) && c.submitWindow() > 1 {
|
|
if submitSlots == nil {
|
|
submitSlots = make(chan struct{}, c.submitWindow())
|
|
}
|
|
select {
|
|
case submitSlots <- struct{}{}:
|
|
case <-c.exceed:
|
|
return
|
|
case <-fatal:
|
|
return
|
|
}
|
|
submitGroup.Add(1)
|
|
go func(response *Response) {
|
|
defer submitGroup.Done()
|
|
defer func() { <-submitSlots }()
|
|
if handleErr := c.handlePacket(response); handleErr != nil {
|
|
select {
|
|
case fatal <- handleErr:
|
|
default:
|
|
}
|
|
}
|
|
}(r)
|
|
continue
|
|
}
|
|
if err = c.handlePacket(r); err != nil {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *conn) submitWindow() int {
|
|
if c.server.SubmitWindow == nil {
|
|
return 1
|
|
}
|
|
window := c.server.SubmitWindow(c.Conn)
|
|
if window < 1 {
|
|
return 1
|
|
}
|
|
if window > 1024 {
|
|
return 1024
|
|
}
|
|
return window
|
|
}
|
|
|
|
func isSubmitPacket(packet Packer) bool {
|
|
switch packet.(type) {
|
|
case *Cmpp2SubmitReqPkt, *Cmpp3SubmitReqPkt:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (c *conn) handlePacket(r *Response) error {
|
|
_, handlerErr := c.server.Handler.ServeCmpp(r, r.Packet, c.server.ErrorLog)
|
|
sendErr := c.finishPacket(r)
|
|
if r.AfterSend != nil {
|
|
r.AfterSend(sendErr)
|
|
}
|
|
if sendErr != nil {
|
|
c.server.ErrorLog.Printf(
|
|
"send response packet failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v",
|
|
c.Conn.RemoteAddr(), c.Conn.Typ, r.Packer, r.SeqId, sendErr, sendErr,
|
|
)
|
|
return sendErr
|
|
}
|
|
if handlerErr != nil {
|
|
c.server.ErrorLog.Printf(
|
|
"handler failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v",
|
|
c.Conn.RemoteAddr(), c.Conn.Typ, r.Packet.Packer, r.SeqId, handlerErr, handlerErr,
|
|
)
|
|
}
|
|
return handlerErr
|
|
}
|
|
|
|
// Create new connection from rwc.
|
|
func (srv *Server) newConn(rwc net.Conn) (c *conn, err error) {
|
|
c = new(conn)
|
|
c.server = srv
|
|
c.Conn = NewConn(rwc, srv.Typ)
|
|
c.Conn.SetState(CONN_CONNECTED)
|
|
c.n = c.server.N
|
|
c.t = c.server.T
|
|
return c, nil
|
|
}
|
|
|
|
func (srv *Server) listenAndServe() error {
|
|
if srv.Addr == "" {
|
|
return ErrEmptyServerAddr
|
|
}
|
|
ln, err := net.Listen("tcp", srv.Addr)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return srv.Serve(tcpKeepAliveListener{ln.(*net.TCPListener)})
|
|
}
|
|
|
|
// ListenAndServe listens on the TCP network address addr
|
|
// and then calls Serve with handler to handle requests.
|
|
func ListenAndServe(addr string, typ Type, t time.Duration, n int32, logWriter io.Writer, handlers ...Handler) error {
|
|
return ListenAndServeWithClose(addr, typ, t, n, logWriter, nil, handlers...)
|
|
}
|
|
|
|
// ListenAndServeWithClose behaves like ListenAndServe and invokes onClose once
|
|
// after an accepted client connection ends, including abrupt TCP disconnects.
|
|
func ListenAndServeWithClose(addr string, typ Type, t time.Duration, n int32, logWriter io.Writer, onClose func(*Conn), handlers ...Handler) error {
|
|
return ListenAndServeWithCloseAndSubmitWindow(addr, typ, t, n, logWriter, onClose, nil, handlers...)
|
|
}
|
|
|
|
// ListenAndServeWithCloseAndSubmitWindow adds bounded per-connection Submit
|
|
// concurrency while keeping login, heartbeat and acknowledgement handling serial.
|
|
func ListenAndServeWithCloseAndSubmitWindow(addr string, typ Type, t time.Duration, n int32, logWriter io.Writer, onClose func(*Conn), submitWindow func(*Conn) int, handlers ...Handler) error {
|
|
if addr == "" {
|
|
return ErrEmptyServerAddr
|
|
}
|
|
|
|
if handlers == nil {
|
|
return ErrNoHandlers
|
|
}
|
|
|
|
var handler Handler
|
|
handler = HandlerFunc(func(r *Response, p *Packet, l *log.Logger) (bool, error) {
|
|
for _, h := range handlers {
|
|
next, err := h.ServeCmpp(r, p, l)
|
|
if err != nil || !next {
|
|
return next, err
|
|
}
|
|
}
|
|
return false, nil
|
|
})
|
|
|
|
if logWriter == nil {
|
|
logWriter = os.Stderr
|
|
}
|
|
server := &Server{Addr: addr, Handler: handler, Typ: typ,
|
|
T: t, N: n,
|
|
ErrorLog: log.New(logWriter, "cmppserver: ", log.LstdFlags), OnClose: onClose, SubmitWindow: submitWindow}
|
|
return server.listenAndServe()
|
|
}
|
|
|
|
// tcpKeepAliveListener sets TCP keep-alive timeouts on accepted
|
|
// connections. It's used by ListenAndServe so
|
|
// dead TCP connections (e.g. closing laptop mid-download) eventually
|
|
// go away. the tcpKeepAliveListener's implementation is copied from
|
|
// http package.
|
|
type tcpKeepAliveListener struct {
|
|
*net.TCPListener
|
|
}
|
|
|
|
func (ln tcpKeepAliveListener) Accept() (c net.Conn, err error) {
|
|
tc, err := ln.AcceptTCP()
|
|
if err != nil {
|
|
return
|
|
}
|
|
tc.SetKeepAlive(true)
|
|
tc.SetKeepAlivePeriod(1 * time.Minute) // 1min
|
|
return tc, nil
|
|
}
|