begin adding NTP client node

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
+}

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() {

ntp/client.go

@@ -27,9 +27,10 @@ func NewClient(now func() time.Time) *Client {
 }
 
 type Client struct {
-	start time.Time
+	connID uint64
-	_now  func() time.Time
+	start  time.Time
-	t     [4]Timestamp
+	_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

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")

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 {
+// IsTxOver returns true if there is no more frames to encapsulate over the network.
-	return h.scb.State().IsClosed()
+// 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 {