// Copyright (c) 2023 The Gnet Authors. All rights reserved.

//

// 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,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

package gnet

import (

"errors"

"io"

"net"

"os"

"syscall"

"time"

"golang.org/x/sys/windows"

"github.com/panjf2000/gnet/v2/pkg/buffer/elastic"

errorx "github.com/panjf2000/gnet/v2/pkg/errors"

bbPool "github.com/panjf2000/gnet/v2/pkg/pool/bytebuffer"

bsPool "github.com/panjf2000/gnet/v2/pkg/pool/byteslice"

"github.com/panjf2000/gnet/v2/pkg/pool/goroutine"

)

type netErr struct {

c *conn

err error

}

type tcpConn struct {

c *conn

b *bbPool.ByteBuffer

}

type udpConn struct {

c *conn

}

type openConn struct {

c *conn

cb func()

}

type conn struct {

pc net.PacketConn

ctx any // user-defined context

loop *eventloop // owner event-loop

buffer *bbPool.ByteBuffer // reuse memory of inbound data as a temporary buffer

cache []byte // temporary cache for the inbound data

rawConn net.Conn // original connection

localAddr net.Addr // local server addr

remoteAddr net.Addr // remote addr

inboundBuffer elastic.RingBuffer // buffer for data from the remote

}

func packTCPConn(c *conn, buf []byte) *tcpConn {

b := bbPool.Get()

_, _ = b.Write(buf)

return &tcpConn{c: c, b: b}

}

func unpackTCPConn(tc *tcpConn) *conn {

if tc.c.buffer == nil { // the connection has been closed

return nil

}

_, _ = tc.c.buffer.Write(tc.b.B)

bbPool.Put(tc.b)

tc.b = nil

return tc.c

}

func packUDPConn(c *conn, buf []byte) *udpConn {

_, _ = c.buffer.Write(buf)

return &udpConn{c}

}

func newStreamConn(el *eventloop, nc net.Conn, ctx any) (c *conn) {

return &conn{

ctx: ctx,

loop: el,

buffer: bbPool.Get(),

rawConn: nc,

localAddr: nc.LocalAddr(),

remoteAddr: nc.RemoteAddr(),

}

}

func (c *conn) release() {

c.ctx = nil

c.localAddr = nil

if c.rawConn != nil {

c.rawConn = nil

c.remoteAddr = nil

}

c.inboundBuffer.Done()

bbPool.Put(c.buffer)

c.buffer = nil

}

func newUDPConn(el *eventloop, pc net.PacketConn, rc net.Conn, localAddr, remoteAddr net.Addr, ctx any) *conn {

return &conn{

ctx: ctx,

pc: pc,

rawConn: rc,

loop: el,

buffer: bbPool.Get(),

localAddr: localAddr,

remoteAddr: remoteAddr,

}

}

func (c *conn) resetBuffer() {

c.buffer.Reset()

c.inboundBuffer.Reset()

c.inboundBuffer.Done()

}

func (c *conn) Read(p []byte) (n int, err error) {

if c.inboundBuffer.IsEmpty() {

n = copy(p, c.buffer.B)

c.buffer.B = c.buffer.B[n:]

if n == 0 && len(p) > 0 {

err = io.ErrShortBuffer

}

return

}

n, _ = c.inboundBuffer.Read(p)

if n == len(p) {

return

}

m := copy(p[n:], c.buffer.B)

n += m

c.buffer.B = c.buffer.B[m:]

return

}

func (c *conn) Next(n int) (buf []byte, err error) {

inBufferLen := c.inboundBuffer.Buffered()

if totalLen := inBufferLen + c.buffer.Len(); n > totalLen {

return nil, io.ErrShortBuffer

} else if n <= 0 {

n = totalLen

}

if c.inboundBuffer.IsEmpty() {

buf = c.buffer.B[:n]

c.buffer.B = c.buffer.B[n:]

return

}

buf = bsPool.Get(n)

_, err = c.Read(buf)

return

}

func (c *conn) Peek(n int) (buf []byte, err error) {

inBufferLen := c.inboundBuffer.Buffered()

if totalLen := inBufferLen + c.buffer.Len(); n > totalLen {

return nil, io.ErrShortBuffer

} else if n <= 0 {

n = totalLen

}

if c.inboundBuffer.IsEmpty() {

return c.buffer.B[:n], err

}

head, tail := c.inboundBuffer.Peek(n)

if len(head) == n {

return head, err

}

buf = bsPool.Get(n)[:0]

buf = append(buf, head...)

buf = append(buf, tail...)

if inBufferLen >= n {

return

}

remaining := n - inBufferLen

buf = append(buf, c.buffer.B[:remaining]...)

c.cache = buf

return

}

func (c *conn) Discard(n int) (int, error) {

if len(c.cache) > 0 {

bsPool.Put(c.cache)

c.cache = nil

}

inBufferLen := c.inboundBuffer.Buffered()

if totalLen := inBufferLen + c.buffer.Len(); n >= totalLen || n <= 0 {

c.resetBuffer()

return totalLen, nil

}

if c.inboundBuffer.IsEmpty() {

c.buffer.B = c.buffer.B[n:]

return n, nil

}

discarded, _ := c.inboundBuffer.Discard(n)

if discarded < inBufferLen {

return discarded, nil

}

remaining := n - inBufferLen

c.buffer.B = c.buffer.B[remaining:]

return n, nil

}

func (c *conn) Write(p []byte) (int, error) {

if c.rawConn == nil && c.pc == nil {

return 0, net.ErrClosed

}

if c.rawConn != nil {

return c.rawConn.Write(p)

}

return c.pc.WriteTo(p, c.remoteAddr)

}

func (c *conn) SendTo(p []byte, addr net.Addr) (int, error) {

if c.pc == nil {

return 0, errorx.ErrUnsupportedOp

}

if addr == nil {

return 0, errorx.ErrInvalidNetworkAddress

}

return c.pc.WriteTo(p, addr)

}

func (c *conn) Writev(bs [][]byte) (int, error) {

if c.pc != nil { // not available for UDP

return 0, errorx.ErrUnsupportedOp

}

if c.rawConn != nil {

bb := bbPool.Get()

defer bbPool.Put(bb)

for i := range bs {

_, _ = bb.Write(bs[i])

}

return c.rawConn.Write(bb.Bytes())

}

return 0, net.ErrClosed

}

func (c *conn) ReadFrom(r io.Reader) (int64, error) {

if c.rawConn != nil {

return io.Copy(c.rawConn, r)

}

return 0, net.ErrClosed

}

func (c *conn) WriteTo(w io.Writer) (n int64, err error) {

if !c.inboundBuffer.IsEmpty() {

if n, err = c.inboundBuffer.WriteTo(w); err != nil {

return

}

}

if c.buffer == nil {

return 0, nil

}

defer c.buffer.Reset()

return c.buffer.WriteTo(w)

}

func (c *conn) Flush() error {

return nil

}

func (c *conn) InboundBuffered() int {

if c.buffer == nil {

return 0

}

return c.inboundBuffer.Buffered() + c.buffer.Len()

}

func (c *conn) OutboundBuffered() int {

return 0

}

func (c *conn) Context() any { return c.ctx }

func (c *conn) SetContext(ctx any) { c.ctx = ctx }

func (c *conn) LocalAddr() net.Addr { return c.localAddr }

func (c *conn) RemoteAddr() net.Addr { return c.remoteAddr }

func (c *conn) Fd() (fd int) {

if c.rawConn == nil {

return -1

}

rc, err := c.rawConn.(syscall.Conn).SyscallConn()

if err != nil {

return -1

}

if err := rc.Control(func(i uintptr) {

fd = int(i)

}); err != nil {

return -1

}

return

}

func (c *conn) Dup() (fd int, err error) {

if c.rawConn == nil && c.pc == nil {

return -1, net.ErrClosed

}

var (

sc syscall.Conn

ok bool

)

if c.rawConn != nil {

sc, ok = c.rawConn.(syscall.Conn)

} else {

sc, ok = c.pc.(syscall.Conn)

}

if !ok {

return -1, errors.New("failed to convert net.Conn to syscall.Conn")

}

rc, err := sc.SyscallConn()

if err != nil {

return -1, errors.New("failed to get syscall.RawConn from net.Conn")

}

var dupHandle windows.Handle

e := rc.Control(func(fd uintptr) {

process := windows.CurrentProcess()

err = windows.DuplicateHandle(

process,

windows.Handle(fd),

process,

&dupHandle,

0,

true,

windows.DUPLICATE_SAME_ACCESS,

)

})

if err != nil {

return -1, err

}

if e != nil {

return -1, e

}

return int(dupHandle), nil

}

func (c *conn) SetReadBuffer(bytes int) error {

if c.rawConn == nil && c.pc == nil {

return net.ErrClosed

}

if c.rawConn != nil {

return c.rawConn.(interface{ SetReadBuffer(int) error }).SetReadBuffer(bytes)

}

return c.pc.(interface{ SetReadBuffer(int) error }).SetReadBuffer(bytes)

}

func (c *conn) SetWriteBuffer(bytes int) error {

if c.rawConn == nil && c.pc == nil {

return net.ErrClosed

}

if c.rawConn != nil {

return c.rawConn.(interface{ SetWriteBuffer(int) error }).SetWriteBuffer(bytes)

}

return c.pc.(interface{ SetWriteBuffer(int) error }).SetWriteBuffer(bytes)

}

func (c *conn) SetLinger(sec int) error {

if c.rawConn == nil {

return net.ErrClosed

}

tc, ok := c.rawConn.(*net.TCPConn)

if !ok {

return errorx.ErrUnsupportedOp

}

return tc.SetLinger(sec)

}

func (c *conn) SetNoDelay(noDelay bool) error {

if c.rawConn == nil {

return net.ErrClosed

}

tc, ok := c.rawConn.(*net.TCPConn)

if !ok {

return errorx.ErrUnsupportedOp

}

return tc.SetNoDelay(noDelay)

}

func (c *conn) SetKeepAlivePeriod(d time.Duration) error {

return c.SetKeepAlive(d > 0, d, d/5, 5)

}

func (c *conn) SetKeepAlive(enabled bool, idle, intvl time.Duration, cnt int) error {

if c.rawConn == nil && c.pc == nil {

return net.ErrClosed

}

if c.pc != nil {

return errorx.ErrUnsupportedOp

}

tc, ok := c.rawConn.(*net.TCPConn)

if !ok {

return errorx.ErrUnsupportedOp

}

if enabled && (idle <= 0 || intvl <= 0 || cnt <= 0) {

return errors.New("invalid time duration")

}

if err := tc.SetKeepAlive(enabled); err != nil {

return err

}

if !enabled {

return nil

}

if err := tc.SetKeepAlivePeriod(idle); err != nil {

return err

}

if err := windows.SetsockoptInt(

windows.Handle(c.Fd()),

windows.IPPROTO_TCP,

windows.TCP_KEEPINTVL,

int(intvl.Seconds())); err != nil {

return os.NewSyscallError("setsockopt", err)

}

if err := windows.SetsockoptInt(

windows.Handle(c.Fd()),

windows.IPPROTO_TCP,

windows.TCP_KEEPCNT,

cnt); err != nil {

return os.NewSyscallError("setsockopt", err)

}

return nil

}

// Gfd return an uninitialized GFD which is not valid,

// this method is only implemented for compatibility, don't use it on Windows.

// func (c *conn) Gfd() gfd.GFD { return gfd.GFD{} }

func (c *conn) AsyncWrite(buf []byte, cb AsyncCallback) error {

fn := func() error {

_, err := c.Write(buf)

if cb != nil {

_ = cb(c, err)

}

return err

}

var err error

select {

case c.loop.ch <- fn:

default:

// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.

err = goroutine.DefaultWorkerPool.Submit(func() {

c.loop.ch <- fn

})

}

return err

}

func (c *conn) AsyncWritev(bs [][]byte, cb AsyncCallback) error {

if c.pc != nil {

return errorx.ErrUnsupportedOp

}

buf := bbPool.Get()

for _, b := range bs {

_, _ = buf.Write(b)

}

return c.AsyncWrite(buf.Bytes(), func(c Conn, err error) error {

defer bbPool.Put(buf)

if cb == nil {

return err

}

return cb(c, err)

})

}

func (c *conn) Wake(cb AsyncCallback) (err error) {

wakeFn := func() (err error) {

err = c.loop.wake(c)

if cb != nil {

_ = cb(c, err)

}

return

}

select {

case c.loop.ch <- wakeFn:

default:

// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.

err = goroutine.DefaultWorkerPool.Submit(func() {

c.loop.ch <- wakeFn

})

}

return

}

func (c *conn) Close() (err error) {

closeFn := func() error {

return c.loop.close(c, nil)

}

select {

case c.loop.ch <- closeFn:

default:

// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.

err = goroutine.DefaultWorkerPool.Submit(func() {

c.loop.ch <- closeFn

})

}

return

}

func (c *conn) CloseWithCallback(cb AsyncCallback) (err error) {

closeFn := func() (err error) {

err = c.loop.close(c, nil)

if cb != nil {

_ = cb(c, err)

}

return

}

select {

case c.loop.ch <- closeFn:

default:

// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.

err = goroutine.DefaultWorkerPool.Submit(func() {

c.loop.ch <- closeFn

})

}

return

}

func (c *conn) EventLoop() EventLoop {

return c.loop

}

func (*conn) SetDeadline(_ time.Time) error {

return errorx.ErrUnsupportedOp

}

func (*conn) SetReadDeadline(_ time.Time) error {

return errorx.ErrUnsupportedOp

}

func (*conn) SetWriteDeadline(_ time.Time) error {

return errorx.ErrUnsupportedOp

}