type Server struct {
    gnet.BuiltinEventEngine
    eng     gnet.Engine
    addr    string
    network string

    bufferSize int
    workerPool *goroutine.Pool

    codec   codec.Codec
    handler Handler
}

// NewServer returns a new Server
func NewServer() *Server {
    return &Server{
        handler:    NewDefaultHandler(),
        codec:      AutomaticCodec,
        workerPool: goroutine.Default(),
    }
}

// SetHandler Sets the handler, this handler with receive every syslog entry
func (s *Server) SetHandler(handler Handler) {
    s.handler = handler
}

func (s *Server) Boot() error {
    err := gnet.Run(s, s.addr,
        gnet.WithMulticore(true),
        gnet.WithSocketRecvBuffer(s.bufferSize))
    if err != nil {
        return err
    }

    return nil
}

func (s *Server) Stop() error {
    _ = s.eng.Stop(context.Background())
    s.workerPool.Release()

    return nil
}

func (s *Server) OnBoot(eng gnet.Engine) gnet.Action {
    s.eng = eng

    logging.Infof("syslog server is listening on %s\n", s.addr)

    return gnet.None
}

func (s *Server) OnTraffic(conn gnet.Conn) (action gnet.Action) {
    if s.network == "udp" {
        return s.handleUdp(conn)
    }

    if s.network == "tcp" {
        return s.handleTcp(conn)
    }

    if s.network == "unix" {
        return s.handleUdp(conn)
    }

    return gnet.None
}

func (s *Server) handleUdp(conn gnet.Conn) (action gnet.Action) {
    data, err := conn.Next(-1)
    if err != nil {
        logging.Errorf("syslog read buff, something wrong, error:%v", err)
        return gnet.None
    }

    client := conn.RemoteAddr().String()
    copyData := make([]byte, len(data))
    copy(copyData, data)
    _ = s.workerPool.Submit(func() {
        p := s.codec.GetParser(copyData)
        log, _ := p.Parse(copyData, client)
        s.handler.Handle(log)
    })

    return gnet.None
}

func (s *Server) handleTcp(conn gnet.Conn) (action gnet.Action) {
    for {
        data, p, err := s.codec.Decode(conn)
        if err != nil {
            break
        }

        client := conn.RemoteAddr().String()
        _ = s.workerPool.Submit(func() {
            log, _ := p.Parse(data, client)
            s.handler.Handle(log)
        })

        return gnet.None
    }

    if conn.InboundBuffered() > 0 {
        if err := conn.Wake(nil); err != nil { // wake up the connection manually to avoid missing the leftover data
            logging.Errorf("failed to wake up the connection, %v", err)
            return gnet.Close
        }
    }

    return gnet.None
}

type Codec interface {
    // Decode 用于 TCP 拆包
    Decode(conn gnet.Conn) ([]byte, parser.Parser, error)
    // GetParser 获取解析器，用于格式解析
    GetParser([]byte) parser.Parser
}

// rfc3164
type RFC3164Codec struct{}

func (f *RFC3164Codec) GetParser(data []byte) parser.Parser {
    return rfc3164Parser
}

func (f *RFC3164Codec) Decode(conn gnet.Conn) ([]byte, parser.Parser, error) {
    buf, _ := conn.Next(-1)

    length := len(buf)
    body := make([]byte, length)
    copy(body, buf)

    _, _ = conn.Discard(length)

    return body, rfc3164Parser, nil
}

// rfc5424
type RFC5424Codec struct{}

func (f *RFC5424Codec) GetParser(data []byte) parser.Parser {
    return rfc5424Parser
}

func (f *RFC5424Codec) Decode(conn gnet.Conn) ([]byte, parser.Parser, error) {
    buf, _ := conn.Next(-1)

    length := len(buf)
    body := make([]byte, length)
    copy(body, buf)

    _, _ = conn.Discard(length)

    return body, rfc5424Parser, nil
}


// rfc6587
type RFC6587Codec struct{}

func (f *RFC6587Codec) GetParser(data []byte) parser.Parser {
    return rfc6587Parser
}

func (f *RFC6587Codec) Decode(conn gnet.Conn) ([]byte, parser.Parser, error) {
    if conn.InboundBuffered() < 4 {
        // 如果没有找到换行符，说明数据不完整，等待更多数据
        return nil, rfc6587Parser, ErrIncompletePacket
    }

    lenBuf, _ := conn.Peek(4)
    length := binary.BigEndian.Uint32(lenBuf)

    _, _ = conn.Discard(4)

    buf, _ := conn.Next(int(length))
    body := make([]byte, length)
    copy(body, buf)

    _, _ = conn.Discard(int(length))

    return nil, rfc6587Parser, nil
}

type Parser interface {
    Parse(data []byte, client string) (*Log, error)
    Location(*time.Location)
}

type Log struct {
    // 结构化的数据
    Header map[string]interface{} `json:"header"`
    // 原始数据
    Body []byte `json:"body"`

    // 是否有错，有错时结构化数据不可能
    Err error

    // 辅助解析用，解析的过程中cursor会发生变化
    cursor int
    // 辅助解析用
    len int
    // 是否忽略标签
    skipTag bool
}

// Handler The handler receive every syslog entry at Handle method
type Handler interface {
    Handle(log *parser.Log)
}

type DefaultHandler struct {
    counter *int64
}

func NewDefaultHandler() *DefaultHandler {
    return &DefaultHandler{
        counter: new(int64),
    }
}

// Handle entry receiver
func (h *DefaultHandler) Handle(log *parser.Log) {
    atomic.AddInt64(h.counter, 1)

    logging.Infof("number %d, header:%v, body:%v,", *h.counter, log.Header, log.GetString(parser.LogBody))
}

func Test_udp_server(t *testing.T) {
    server := NewServer()
    server.SetCodec(rfc3164Codec)
    server.SetAddr("udp://0.0.0.0:514")
    defer func(server *Server) {
        _ = server.Stop()
    }(server)

    _ = server.Boot()
}