diff --git a/internet/node-ntp.go b/internet/node-ntp.go new file mode 100644 index 0000000..2b03565 --- /dev/null +++ b/internet/node-ntp.go @@ -0,0 +1,31 @@ +package internet + +import ( + "github.com/soypat/lneto/ntp" +) + +var _ StackNode = (*NodeNTPClient)(nil) + +type NodeNTPClient struct { + c ntp.Client +} + +func (n *NodeNTPClient) Protocol() uint64 { + return 0 +} + +func (n *NodeNTPClient) LocalPort() uint16 { + return ntp.ClientPort +} + +func (n *NodeNTPClient) ConnectionID() *uint64 { + return n.c.ConnectionID() +} + +func (n *NodeNTPClient) Demux(carrierData []byte, ntpOffset int) error { + return nil +} + +func (n *NodeNTPClient) Encapsulate(carrierData []byte, ntpOffset int) (int, error) { + return 0, nil +} diff --git a/internet/node-tcplistener.go b/internet/node-tcplistener.go index 35ee0ba..e43c37a 100644 --- a/internet/node-tcplistener.go +++ b/internet/node-tcplistener.go @@ -5,7 +5,6 @@ import ( "errors" "log/slog" "net" - "time" "github.com/soypat/lneto" "github.com/soypat/lneto/internal" @@ -94,30 +93,6 @@ func (listener *NodeTCPListener) TryAccept() (*tcp.Conn, error) { return nil, errors.New("no conns available") } -func (listener *NodeTCPListener) AcceptRaw() (*tcp.Conn, error) { - connid := listener.connID - for { - if listener.isClosed() || connid != listener.connID { - return nil, net.ErrClosed - } - for i, conn := range listener.ready { // Scan ready to see if we can accept. - if conn == nil { - continue - } - state := conn.State() - if state != tcp.StateEstablished { - continue // Do not accept until established. - } - listener.accepted = append(listener.accepted, conn) - listener.ready[i] = nil // discard from ready. - return conn, nil - } - listener.maintainConns() - time.Sleep(5 * time.Millisecond) - } - panic("unreachable") -} - // Encapsulate implements [StackNode]. func (listener *NodeTCPListener) Encapsulate(carrierData []byte, tcpFrameOffset int) (int, error) { if listener.isClosed() { diff --git a/ntp/client.go b/ntp/client.go index 593146a..d63f953 100644 --- a/ntp/client.go +++ b/ntp/client.go @@ -27,9 +27,10 @@ func NewClient(now func() time.Time) *Client { } type Client struct { - start time.Time - _now func() time.Time - t [4]Timestamp + connID uint64 + start time.Time + _now func() time.Time + t [4]Timestamp // org Timestamp // rec Timestamp xmt Timestamp @@ -38,6 +39,21 @@ type Client struct { _sysprec int8 } +func (c *Client) Reset(now func() time.Time) { + if c._sysprec == 0 { + c._sysprec = sysprecRecalcNeeded + } + *c = Client{ + connID: c.connID + 1, + _now: now, + _sysprec: c._sysprec, + } +} + +func (c *Client) ConnectionID() *uint64 { + return &c.connID +} + func (c *Client) Send(payload []byte) (int, error) { if c.isDone() { return 0, io.EOF diff --git a/tcp/conn.go b/tcp/conn.go index 1b0d621..4558072 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -112,29 +112,6 @@ func (conn *Conn) Abort() { } } -func (conn *Conn) Demux(buf []byte, off int) (err error) { - conn.trace("tcpconn.Recv:start") - if off >= len(buf) { - return errors.New("bad offset in TCPConn.Recv") - } - raddr, id, err := internal.GetIPSourceAddr(buf[:off]) - if err != nil { - return err - } - if conn.isRaddrSet() && !bytes.Equal(conn.remoteAddr, raddr) { - return errors.New("IP addr mismatch on TCPConn") - } - err = conn.h.Recv(buf[off:]) - if err != nil { - return err - } - if !conn.isRaddrSet() && conn.h.RemotePort() != 0 { - conn.remoteAddr = append(conn.remoteAddr[:0], raddr...) - conn.ipID = ^(id - 1) - } - return nil -} - // Write writes argument data to the TCPConns's output buffer which is queued to be sent. func (conn *Conn) Write(b []byte) (int, error) { err := conn.checkPipeOpen() @@ -212,6 +189,29 @@ func (conn *Conn) checkPipeOpen() error { return nil } +func (conn *Conn) Demux(buf []byte, off int) (err error) { + conn.trace("tcpconn.Recv:start") + if off >= len(buf) { + return errors.New("bad offset in TCPConn.Recv") + } + raddr, id, err := internal.GetIPSourceAddr(buf[:off]) + if err != nil { + return err + } + if conn.isRaddrSet() && !bytes.Equal(conn.remoteAddr, raddr) { + return errors.New("IP addr mismatch on TCPConn") + } + err = conn.h.Recv(buf[off:]) + if err != nil { + return err + } + if !conn.isRaddrSet() && conn.h.RemotePort() != 0 { + conn.remoteAddr = append(conn.remoteAddr[:0], raddr...) + conn.ipID = ^(id - 1) + } + return nil +} + func (conn *Conn) Encapsulate(buf []byte, off int) (n int, err error) { if len(conn.remoteAddr) == 0 { return 0, errors.New("unset IP address") diff --git a/tcp/handler.go b/tcp/handler.go index 01bcb5e..10c31f9 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -136,7 +136,7 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) { // Recv receives an incoming TCP packet frame with the first byte being the first octet of the TCP frame. // The [Handler]'s internal state is updated if the packet is admitted successfully. func (h *Handler) Recv(incomingPacket []byte) error { - if h.isClosed() { + if h.IsTxOver() { return net.ErrClosed } tfrm, err := NewFrame(incomingPacket) @@ -211,7 +211,7 @@ func (h *Handler) Close() error { // The returned integer is the length written to the argument buffer. func (h *Handler) Send(b []byte) (int, error) { h.trace("tcp.Handler:start", slog.Uint64("port", uint64(h.localPort))) - if h.State().IsClosed() && !h.AwaitingSynSend() { + if h.IsTxOver() { return 0, net.ErrClosed } tfrm, err := NewFrame(b) @@ -302,11 +302,6 @@ func (h *Handler) BufferedInput() int { return h.bufRx.Buffered() } -// InUse returns true if the connection has been initialized and is being used to reach a remote port or if it is awaiting a remote packet. -func (h *Handler) InUse() bool { - return h.remotePort != 0 || !h.State().IsClosed() -} - // AwaitingSynResponse returns true if the Handler is an active client opened with [Handler.OpenActive] and has already sent out the first SYN packet to the remote client. func (h *Handler) AwaitingSynResponse() bool { return h.remotePort != 0 && h.scb.State() == StateSynSent @@ -322,8 +317,12 @@ func (h *Handler) AwaitingSynSend() bool { return h.remotePort != 0 && h.scb.State() == StateClosed } -func (h *Handler) isClosed() bool { - return h.scb.State().IsClosed() +// IsTxOver returns true if there is no more frames to encapsulate over the network. +// The connection is pretty much over in this case if packets made it succesfully to remote. +func (h *Handler) IsTxOver() bool { + state := h.State() + return state == StateClosed && !h.AwaitingSynSend() || + state == StateTimeWait && !h.scb.HasPending() } func min(a, b int) int {