//kite: file: handler.go

//Implement LoadMorePerUserInbox Method.
func LoadMorePerUserInbox(ctx context.Context, r *message_api.PerUserMessagesRequest) (response *message_api.PerUserMessagesResponse, err error) {
    //Implement your method here
    return nil, nil
}

//or gRPC: 
func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, stream pb.RouteGuide_ListFeaturesServer) error {
        ...
}
...

int main(int argc , char *argv[])
{
    //Create struct for server addr and client addr
    struct sockaddr_in cliaddr, servaddr;
    //Create socket FD for TCP
    listenfd = socket(AF_INET , SOCK_STREAM , 0);
    //Prepare the sockaddr_in structure
    server.sin_family = AF_INET;
    server.sin_addr.s_addr = INADDR_ANY;
    server.sin_port = htons( 8888 );
    //Bind
    bind(listenfd, (struct sockaddr *)&servaddr , sizeof(server));
    //Listen
    listen(socket_desc , 3);    
    //Accept
    for ( ; ; ){
        c = sizeof(struct sockaddr_in);
        //Get a new socketfd from the listener queue
        connfd = accept(socket_desc, (struct sockaddr *)&client, (socklen_t*)&c);
        //Handle with connfd
        //e.g. n = recv(connfd, buf, MAXLINE,0)
    }
    //Other Logics
    return 0;
}

for ( ; ; ){
    c = sizeof(struct sockaddr_in);
    //Get a new socketfd from the 
    connfd = accept(socket_desc, (struct sockaddr *)&client, (socklen_t*)&c);
    //Block here and await receive returns the data buffer
    int recvlen = recv(connfd, buf, RECV_BUF_SIZE,0)
}

// Change Socket to Non Blocking
int main(int argc , char *argv[])
{
    //...
    //Create socket FD
    listenfd = socket(AF_INET , SOCK_STREAM , 0);
    //Change to Non Blocking Sample
    fcntl(sock_fd, F_SETFL, fdflags | O_NONBLOCK);
    //Loop call
    while(1)  {  
        int recvlen = recv(sock_fd, recvbuf, RECV_BUF_SIZE) ; 
        ......
    }
    //When system returns error like connection failed/terminated
    close(connfd); 
    close(listenfd)
    //Other Logics
    return 0;
}

//return num of ready fds
int select (int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

typedef union epoll_data {
    int      fd;
    //...
} epoll_data_t;

struct epoll_event {
    uint32_t     events;    /* Epoll events */
    epoll_data_t data;      /* User data variable */
};
//Core functions
int epoll_create(int size);
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout); // ready events will be stored in the array we passed in

package main

import (
        "log"
        "net"
)

func main() {
        //create listener
        listen, err := net.Listen("tcp", ":8888")
        if err != nil {
                log.Println("listen error: ", err)
                return
        }
        for {   //block here to create connection
                conn, err := listen.Accept()
                if err != nil {
                        log.Println("accept error: ", err)
                        break
                }
                // start a new goroutine to handle the new connection.
                go HandleConn(conn)
        }
}
 
func HandleConn(conn net.Conn) {
        defer conn.Close()
        packet := make([]byte, 1024)
        for {   // block here if socket is not available for reading data.
                n, err := conn.Read(packet)
                if err != nil {
                        log.Println("read socket error: ", err)
                        return
                }
                // same as above, block here if socket is not available for writing.
                _, _ = conn.Write(packet[:n])
        }
}

// TCPListener is a TCP network listener. Clients should typically
// use variables of type Listener instead of assuming TCP.
type TCPListener struct {
        fd *netFD
        lc ListenConfig
}

// TCPConn is an implementation of the Conn interface for TCP network
// connections.
type TCPConn struct {
        conn
}
 
// Conn
type conn struct {
        fd *netFD
}

// Network file descriptor.
type netFD struct {
        pfd poll.FD
 
        // immutable until Close
        family      int
        sotype      int
        isConnected bool // handshake completed or use of association with peer
        net         string
        laddr       Addr
        raddr       Addr
}
 
// FD is a file descriptor. The net and os packages use this type as a
// field of a larger type representing a network connection or OS file.
type FD struct {
        // Lock sysfd and serialize access to Read and Write methods.
        fdmu fdMutex
        // System file descriptor. Immutable until Close.
        Sysfd int
        // I/O poller.
        pd pollDesc
        // Writev cache.
        //...... other fields
}

package main

import (
   "context"
   "time"
   "code.byted.org/middleware/netpoll"
)

func main() {
    //set poller number 
    netpoll.SetNumLoops(5)    
   // Create listener（wrapper on net.Listener）
   listener, err := netpoll.CreateListener("tcp", "127.0.0.1:8888")
   if err != nil {
      panic("create netpoll listener fail")
   }
   // handler assign handler
   var onRequest netpoll.OnRequest = handler

   // options: EventLoop init options
   var opts = []netpoll.Option{
      netpoll.WithReadTimeout(1 * time.Second),
      netpoll.WithIdleTimeout(10 * time.Minute),
      netpoll.WithOnPrepare(nil),
   }

   // Create EventLoop
   eventLoop, err := netpoll.NewEventLoop(onRequest, opts...)
   if err != nil {
      panic("create netpoll event-loop fail")
   }

   // Run Server and pass in the listener 
   err = eventLoop.Serve(listener)
   if err != nil {
      panic("netpoll server exit")
   }
}

// Handler on connection
func handler(ctx context.Context, connection netpoll.Connection) error {
   return connection.Writer().Flush()
}

// FDOperator is a collection of operations on file descriptors.
type FDOperator struct {
   // FD is file descriptor, poll will bind when register.
   FD int

   // The FDOperator provides three operations of reading, writing, and hanging.
   // The poll actively fire the FDOperator when fd changes, no check the return value of FDOperator.
   OnRead  func(p Poll) error //for listener
   OnWrite func(p Poll) error //for listener
   OnHup   func(p Poll) error //for listener

   // The following is the required fn, which must exist when used, or directly panic.
   // Fns are only called by the poll when handles connection events.
   Inputs   func(vs [][]byte) (rs [][]byte)  //for connection
   InputAck func(n int) (err error) //for connection

   // Outputs will locked if len(rs) > 0, which need unlocked by OutputAck.
   Outputs   func(vs [][]byte) (rs [][]byte, supportZeroCopy bool) //for connection
   OutputAck func(n int) (err error) //for connection

   // poll is the registered location of the file descriptor.
   poll Poll 

   // private, used by operatorCache
   next  *FDOperator
   state int32 // CAS: 0(unused) 1(inuse) 2(do-done)
}

// Poll monitors fd(file descriptor), calls the FDOperator to perform specific actions,
// and shields underlying differences. On linux systems, poll uses epoll by default,
// and kevent by default on bsd systems.
type Poll interface {
   // Wait will poll all registered fds, and schedule processing based on the triggered event.
   // The call will block, so the usage can be like:
   //
   //  go wait()
   //
   Wait() error

   // Close the poll and shutdown Wait().
   Close() error

   // Trigger can be used to actively refresh the loop where Wait is located when no event is triggered.
   // On linux systems, eventfd is used by default, and kevent by default on bsd systems.
   Trigger() error

   // Control the event of file descriptor and the operations is defined by PollEvent.
   Control(operator *FDOperator, event PollEvent) error
}

//github.com/cloudwego/[email protected]/poll_default_linux.go
type defaultPoll struct {
   pollArgs
   fd      int         // epoll fd
   wop     *FDOperator // eventfd, wake epoll_wait
   buf     []byte      // read wfd trigger msg
   trigger uint32      // trigger flag
   // fns for handle events
   Reset   func(size, caps int)
   Handler func(events []epollevent) (closed bool)
}

type pollArgs struct {
   size     int
   caps     int
   events   []epollevent
   barriers []barrier
}

func init() {
   var loops = runtime.GOMAXPROCS(0)/20 + 1 //number of poller
   pollmanager = &manager{}
   pollmanager.SetLoadBalance(RoundRobin)
   pollmanager.SetNumLoops(loops)
}

type manager struct {
   NumLoops int
   balance  loadbalance // load balancing method (round robin)
   polls    []Poll      // all the polls
}
func (m *manager) Run() error
func (m *manager) Pick() Poll

// Wait implements Poll.
func (p *defaultPoll) Wait() (err error) {
   // msec is the time the EpollWait need to wait before returns anything
   //0 means instant return
   //-1 means blocking
   //10 means return in 10ms (iirc)
   var caps, msec, n = barriercap, -1, 0
   p.Reset(128, caps)
   // wait
   for {
      if n == p.size && p.size < 128*1024 {
         p.Reset(p.size<<1, caps)
      }
      n, err = EpollWait(p.fd, p.events, msec)
      if err != nil && err != syscall.EINTR {
         return err
      }
      if n <= 0 { 
         //This mean the previous loop n must be > 0
         //So this round no event (LT events finished)
         //Set epollwait to be -1 blocking to reduce syscall
         msec = -1 
         //Actively quit scheduling and delay next EpollWait
         //since next call will most likely be await for new events and blocked
         runtime.Gosched()
         continue
      }
      //This means n > 0 and it means there are events happening
      //Likely next for loop there will also be events (LT)
      //Set epoll wait to zero in speed it up
      
      msec = 0
      if p.Handler(p.events[:n]) {
         return nil
      }
   }
}

func EpollWait(epfd int, events []epollevent, msec int) (n int, err error) {
   var r0 uintptr
   var _p0 = unsafe.Pointer(&events[0])
   if msec == 0 {
      r0, _, err = syscall.RawSyscall6(syscall.SYS_EPOLL_WAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), 0, 0, 0)
   } else {
      r0, _, err = syscall.Syscall6(syscall.SYS_EPOLL_WAIT, uintptr(epfd), uintptr(_p0), uintptr(len(events)), uintptr(msec), 0, 0)
   }
   if err == syscall.Errno(0) {
      err = nil
   }
   return int(r0), err
}

type server struct {
   operator    FDOperator
   ln          Listener
   prepare     OnPrepare // 连接创建后的准备工作（注册到 epoll 中去）
   quit        func(err error)
   connections sync.Map // key=fd, value=connection
}

func (s *server) Run() (err error)
func (s *server) Close(ctx context.Context) error
func (s *server) OnRead(p Poll) error
func (s *server) OnHup(p Poll) error

// Control implements Poll.
func (p *defaultPoll) Control(operator *FDOperator, event PollEvent) error {
   var op int
   var evt epollevent
   //store the FDOperator pointer to the epoll event
   *(**FDOperator)(unsafe.Pointer(&evt.data)) = operator 
   //e.g. readable event
   op, evt.events = syscall.EPOLL_CTL_ADD, syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLERR
   return EpollCtl(p.fd, op, operator.FD, &evt)
}

func (p *defaultPoll) handler(events []epollevent) (closed bool) {
   for i := range events {
      //Get operator
      var operator = *(**FDOperator)(unsafe.Pointer(&events[i].data))
      if !operator.do() {
         continue
      }
      // trigger or exit gracefully
      // skip

      evt := events[i].events
      // check poll in
      if evt&syscall.EPOLLIN != 0 {
          operator.OnRead(p)
      }
      // check hup
      if evt&(syscall.EPOLLHUP|syscall.EPOLLRDHUP) != 0 {
         p.appendHup(operator)
         continue
      } 
      operator.done()
   }
   // hup conns together to avoid blocking the poll.
   p.detaches()
   return false
}

type Connection interface {
   //兼容 API，不推荐使用
   net.Conn
   
   //zero-copy reader & writer
   Reader() Reader
   Writer() Writer

   IsActive() bool
   SetReadTimeout(timeout time.Duration) error
   SetIdleTimeout(timeout time.Duration) error
   SetOnRequest(on OnRequest) error
   //由于存在被动关闭的情况，所以需要注册关闭回调
   AddCloseCallback(callback CloseCallback) error
}

type Reader interface {
   Next(n int) (p []byte, err error)
   Peek(n int) (buf []byte, err error)
   Skip(n int) (err error)
   ReadString(n int) (s string, err error)
   ReadBinary(n int) (p []byte, err error)
   ReadByte() (b byte, err error)
   Slice(n int) (r Reader, err error)
   Release() (err error)
   Len() (length int)
}

type Writer interface {
   Malloc(n int) (buf []byte, err error)
   WriteString(s string) (n int, err error)
   WriteBinary(b []byte) (n int, err error)
   WriteByte(b byte) (err error)
   WriteDirect(p []byte, remainCap int) error
   MallocAck(n int) (err error)
   Append(w Writer) (n int, err error)
   Flush() (err error)
   MallocLen() (length int)
} 

// connection is the implement of Connection
type connection struct {
   netFD
   onEvent
   locker
   operator        *FDOperator
   readTimeout     time.Duration
   readTimer       *time.Timer
   readTrigger     chan struct{}
   waitReadSize    int32
   writeTrigger    chan error
   inputBuffer     *LinkBuffer
   outputBuffer    *LinkBuffer
   inputBarrier    *barrier
   outputBarrier   *barrier
   supportZeroCopy bool
   maxSize         int // The maximum size of data between two Release().
   bookSize        int // The size of data that can be read at once.
}

//synchrounou handler by poller
func (p *defaultPoll) handler(events []epollevent) (closed bool) {
    for i := range events { // loop epoll events returned
        var operator = *(**FDOperator)(unsafe.Pointer(&events[i].data))
        evt := events[i].events             
        switch {
        // ...
        case evt&syscall.EPOLLIN != 0: // epoll read
            // acquire a linked buffer from linked buffer and book for 8kb
            var bs = operator.Inputs(p.barriers[i].bs)
            if len(bs) > 0 {
                var n, err = readv(operator.FD, bs, p.barriers[i].ivs) // syscall
                operator.InputAck(n)
                if err != nil && err != syscall.EAGAIN && err != syscall.EINTR { // read fail and hop
                    hups = append(hups, operator)
                    break
                }
            }
        case evt&syscall.EPOLLOUT != 0: // epoll write
            //..skip outputs and outputsack
        }
    }
}
// inputAck implements FDOperator.
func (c *connection) inputAck(n int) (err error) {
   if n <= 0 {
      c.inputBuffer.bookAck(0) //convert the buffer to be readable
      return nil
   }

   var needTrigger = true
   if length == n { // first start onRequest
      needTrigger = c.onRequest() //if alr started then just trigger next reading
   }
   if needTrigger && length >= int(atomic.LoadInt32(&c.waitReadSize)) {
      c.triggerRead()
   }
   return nil
}

// Handler on connection
func handler(ctx context.Context, connection netpoll.Connection) error {
   //connection.Reader().Next()
   //connection.Reader().Slice()
   return connection.Writer().Flush()
}


// Next implements Connection.
func (c *connection) Next(n int) (p []byte, err error) {
   if err = c.waitRead(n); err != nil {
      return p, err
   }
   return c.inputBuffer.Next(n)
}

type TTransport interface {
        io.ReadWriteCloser
        Flusher

        // Opens the transport for communication
        Open() error// Returns true if the transport is open
        IsOpen() bool
}

//IMPL
type TBufferedTransport struct {
   bufio.ReadWriter //Buffer pool to get golang bufio
   tp TTransport
}

//gopkg/thrift/protocol.go
type TProtocol interface {
        WriteMessageBegin(name string, typeId TMessageType, seqid int32) error// ......
        ReadMessageBegin() (name string, typeId TMessageType, seqid int32, err error)
        WriteStructBegin(name string) error
        WriteStructEnd() error        
        // ......
        Transport() TTransport
}
//IMPL
type TBinaryProtocol struct {
   trans         TRichTransport
   origTransport TTransport
   reader        io.Reader
   writer        io.Writer
   strictRead    bool
   strictWrite   bool
   buffer        [64]byte
}

// A processor is a generic object which operates upon an input stream and
// writes to some output stream.
type TProcessor interface {
   Process(in, out TProtocol) (bool, TException)
}
//im.conversation.api2   -> finally have somethign famliar
type ConversationServiceProcessor struct {
   processorMap map[string]thrift.TProcessorFunctionWithContext
   handler      ConversationService
}

func main(){
    kite.Init()
    kite.Run()
}

func Init() {
        //1.SetMacPROC
        //2.InitConf
        //3.InitMetrics/tracing
        //...
        // create the singleton server instance
        RPCServer = NewRpcServer() // 3
        _, processorSupportContext = Processor.(thrift.TProcessorWithContext)
}

type RpcServer struct {
        l net.Listener
        processorFactory thrift.TProcessorFactory 
        transportFactory thrift.TTransportFactory
        protocolFactory  thrift.TProtocolFactory
        remoteConfiger *remoteConfiger
        overloader     *overloader
}

// Run starts the default RpcServer.
// It blocks until recv SIGTERM, SIGHUP or SIGINT.
func Run() error {
        errCh := make(chan error, 1)
        go func() { errCh <- RPCServer.ListenAndServe() }() //never stop server
        if err := waitSignal(errCh); err != nil {
                return err
        }
        return RPCServer.Stop()
}
// ListenAndServe ...
func (p *RpcServer) ListenAndServe() error {
   l, err := p.CreateListener()
   //...
   return p.Serve(l)
}
// CreateListener .
func (p *RpcServer) CreateListener() (net.Listener, error) {
   var addr string
   //ListenType TCP
   l, err := net.Listen(ListenType, addr)
   //..
   return l, nil
}
// Serve ...
func (p *RpcServer) Serve(ln net.Listener) error {
   if p.l != nil {
      panic("KITE: Listener not nil")
   }
   p.l = ln
   p.processorFactory = thrift.NewTProcessorFactory(Processor) 
   p.startDaemonRoutines()
   //..
   for {
      stream, err := p.l.Accept()
      //..skip
      go p.handleConn(stream, isShmIPC)
   }
}
//
func (p *RpcServer) handleConn(conn net.Conn, isShmIPC bool) {
   // skip panic handler, overloader
   //Call factory to create the processor transport and protocol instances
   ctx, processor, transport, protocol := p.prepareRequests(conn, isShmIPC)
   // close conn
   var closer io.Closer = transport
   defer closer.Close()

   handleRPC := func() {
      if err := p.processRequests(ctx, processor, transport, protocol); err != nil {
        //log
      }
   }
   if processorSupportContext {
      handleRPC()
   } else {
      if enableGLS || GetRealIP {
         // Use gls to pass context
         glsStorage := gls.NewStorage()
         //handler
      } else {
         handleRPC()
      }
   }
}

//processRequests
if processorSupportContext {
   ok, err = processor.(thrift.TProcessorWithContext).ProcessWithContext(ctx, protocol, protocol)
} else {
   ok, err = processor.Process(protocol, protocol)
}

func init() {
   kite.Processor = message_api.NewImMessageServiceProcessor(&ImMessageServiceHandler{})
}

func NewImMessageServiceProcessor(handler ImMessageService) *ImMessageServiceProcessor {
   //store ur handler name in a processor map
   //We use string because the thrift would encode the handler name as a string
   self123 := &ImMessageServiceProcessor{handler: handler, processorMap: make(map[string]thrift.TProcessorFunctionWithContext)}
   self123.processorMap["SendMessage"] = &imMessageServiceProcessorSendMessage{handler: handler}
   self123.processorMap["GetMessagePerUserInbox"] = &imMessageServiceProcessorGetMessagePerUserInbox{handler: handler}
   self123.processorMap["LoadMorePerUserInbox"] = &imMessageServiceProcessorLoadMorePerUserInbox{handler: handler}
   self123.processorMap["LoadMorePerConversationInbox"] = &imMessageServiceProcessorLoadMorePerConversationInbox{handler: handler}
   //....
}

// TProtocol will find the handler name according to the Protocol u use (Binary)
func (p *ImMessageServiceProcessor) ProcessWithContext(ctx context.Context, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
        // message begin -> method name such as "LoadMorePerUserInbox"
        name, _, seqId, err := iprot.ReadMessageBegin()
        if err != nil {
                return false, err
        }
        if processor, ok := p.GetProcessorFunction(name); ok {
                return processor.Process(ctx, seqId, iprot, oprot)
        }
        // If no handler is found
        iprot.Skip(thrift.STRUCT)
        iprot.ReadMessageEnd()
        x62 := thrift.NewTApplicationException(thrift.UNKNOWN_METHOD, "Unknown function "+name) //famliar right
        oprot.WriteMessageBegin(name, thrift.EXCEPTION, seqId)
        x62.Write(oprot)
        oprot.WriteMessageEnd()
        oprot.Flush()
        return false, x62

}

//code.byted.org/im_cloud/common_tiktok/gen/thrift_gen/message_api/immessageservice.go
func (p *imMessageServiceProcessorSendMessage) Process(ctx context.Context, seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
   args := SendMessageArgs{} //new send msg request
   if err = args.Read(iprot); err != nil { //read
      //handle error
      return false, err
   }
   
   iprot.ReadMessageEnd()
   result := SendMessageResult{}
   var retval *SendMessageResponse
   var err2 error
   if retval, err2 = p.handler.SendMessage(ctx, args.Req); err2 != nil {
      //error
      return true, err2
   } else {
      result.Success = retval
   }
   //oprot.WriteMessageBegin("SendMessage", thrift.REPLY, seqId)
   if err2 = oprot.WriteMessageBegin("SendMessage", thrift.REPLY, seqId)
   if err2 = result.Write(oprot); //..
   if err2 = oprot.WriteMessageEnd();//..
   if err2 = oprot.Flush();//..
   return true, err
}
//Read Struct  
func (p *SendMessageArgs) Read(iprot thrift.TProtocol) error {
   if _, err := iprot.ReadStructBegin(); 
   for {
      _, fieldTypeId, fieldId, err := iprot.ReadFieldBegin()
      if fieldTypeId == thrift.STOP { //finish reading
         break
      }
      switch fieldId {
      case 1:
         if err := p.ReadField1(iprot); 
      default:
         if err := iprot.Skip(fieldTypeId); 
      }
      if err := iprot.ReadFieldEnd(); 
   }
   if err := iprot.ReadStructEnd(); 
   return nil
}