flightsimulatorplatfrom/openfsd
1
0
Fork 0
mirror of https://github.com/renorris/openfsd synced 2026-08-13 04:55:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: internal/session; unify sendError; slog; field ownership docs

Browse Source 
Single Client→Session rename wave. Session owns send channel; no postoffice import.
This commit is contained in:
Reese Norris
2026-07-12 20:15:37 -04:00
parent f74067b6cd
commit 4949789ae7
11 changed files with 590 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
fsd/client.go
View File

@@ -1,129 +0,0 @@
package fsd
import (
"bufio"
"context"
"net"
"github.com/renorris/openfsd/internal/auth"
"github.com/renorris/openfsd/pkg/protocol"
"go.uber.org/atomic"
)
type Client struct {
conn net.Conn
scanner *bufio.Scanner
ctx context.Context
cancelCtx func()
sendChan chan string
coords atomic.Value
visRange atomic.Float64
closestVelocityClientDistance float64 // The closest Velocity-compatible client in meters
flightPlan atomic.String
assignedBeaconCode atomic.String
frequency atomic.String // ATC frequency
altitude atomic.Int32 // Pilot altitude
groundspeed atomic.Int32 // Pilot ground speed
transponder atomic.String // Active pilot transponder
heading atomic.Int32 // Pilot heading
lastUpdated atomic.Time // Last updated time
facilityType int // ATC facility type. This value is only relevant for ATC
loginData
authState auth.AuthState
sendFastEnabled bool
}
type LatLon struct {
lat, lon float64
}
func newClient(ctx context.Context, conn net.Conn, scanner *bufio.Scanner, loginData loginData) (client *Client) {
clientCtx, cancel := context.WithCancel(ctx)
client = &Client{
conn: conn,
scanner: scanner,
ctx: clientCtx,
cancelCtx: cancel,
sendChan: make(chan string, 32),
loginData: loginData,
}
client.setLatLon(0, 0)
return
}
func (c *Client) senderWorker() {
defer c.conn.Close()
defer c.cancelCtx()
for {
select {
case packet := <-c.sendChan:
if _, err := c.conn.Write([]byte(packet)); err != nil {
return
}
case <-c.ctx.Done():
return
}
}
}
// sendError sends an FSD error packet to a Client with the specified code and message.
// It returns an error if writing to the connection fails.
//
// This call is thread-safe
func (c *Client) sendError(code int, message string) (err error) {
return c.send(protocol.FormatError(protocol.ErrorCode(code), message))
}
// send sends a packet string to a Client.
// This call queues the packet in the Client's outbound send channel.
// This call will block until the packet can be queued in the send channel.
// Returns a context error if the Client's context has elapsed.
func (c *Client) send(packet string) (err error) {
select {
case c.sendChan <- packet:
return
case <-c.ctx.Done():
return c.ctx.Err()
}
}
func (s *Server) eventLoop(client *Client) {
defer client.cancelCtx()
go client.senderWorker()
for {
if !client.scanner.Scan() {
return
}
// Reference the next packet
packet := client.scanner.Bytes()
packet = append(packet, '\r', '\n') // Re-append delimiter
// Verify packet and obtain type
packetType, ok := verifyPacket(packet, client)
if !ok {
continue
}
// Run handler
handler := s.getHandler(packetType)
handler(client, packet)
}
}
func (c *Client) latLon() [2]float64 {
latLon := c.coords.Load().(LatLon)
return [2]float64{latLon.lat, latLon.lon}
}
func (c *Client) setLatLon(lat, lon float64) {
c.coords.Store(LatLon{lat: lat, lon: lon})
}

211
fsd/conn.go
View File

@@ -7,40 +7,41 @@ import (
"errors"
"fmt"
"io"
"log/slog"
"net"
"strconv"
"strings"
"time"
"github.com/renorris/openfsd/db"
"github.com/renorris/openfsd/internal/auth"
"github.com/renorris/openfsd/internal/auth"
"github.com/renorris/openfsd/internal/session"
"github.com/renorris/openfsd/pkg/protocol"
)
// sendError sends an FSD error packet to an io.Writer with the specified code and message.
// sendError writes an FSD error packet to an io.Writer (login-phase only).
// It returns an error if writing to the connection fails.
//
// This function must only be used during the login phase,
// as it synchronously writes the error directly to the
// connection socket.
// connection socket. Post-login code must use session.Session.SendError.
func sendError(conn io.Writer, code int, message string) (err error) {
return protocol.WriteError(conn, protocol.ErrorCode(code), message)
}
// handleConn manages a single Client connection.
// If any errors occur during the process, it sends an error to the Client and closes the connection.
// handleConn manages a single client connection.
// If any errors occur during the process, it sends an error to the client and closes the connection.
func (s *Server) handleConn(ctx context.Context, conn net.Conn) {
defer func() {
if err := recover(); err != nil {
fmt.Println("An FSD connection goroutine panicked:")
fmt.Println(err)
slog.Error("FSD connection goroutine panicked", "err", err)
}
}()
defer conn.Close()
if err := sendServerIdent(conn); err != nil {
fmt.Printf("Error sending server ident: %v\n", err)
slog.Debug("error sending server ident", "err", err)
return
}
@@ -54,14 +55,15 @@ func (s *Server) handleConn(ctx context.Context, conn net.Conn) {
}
// Check if the requested callsign is OK
if !isValidClientCallsign([]byte(data.callsign)) {
if !isValidClientCallsign([]byte(data.Callsign)) {
sendError(conn, CallsignInvalidError, "Callsign invalid")
return
}
client := newClient(ctx, conn, scanner, data)
client := session.New(ctx, conn, scanner, data)
client.Auth = &auth.AuthState{}
// Attempt to authenticate connection
// Attempt to authenticate connection (login-phase errors still use sendError on conn)
if err = s.attemptAuthentication(client, token); err != nil {
return
}
@@ -75,8 +77,14 @@ func (s *Server) handleConn(ctx context.Context, conn net.Conn) {
}
defer s.postOffice.release(client)
// Start sender before any post-login outbound traffic (MOTD, etc.).
// After this point, all writes go through client.Send → SenderWorker.
// Direct conn.Write is forbidden outside SenderWorker.
go client.SenderWorker()
// Send hello message to client
if err = s.sendMotd(client); err != nil {
client.Cancel()
return
}
@@ -87,7 +95,33 @@ func (s *Server) handleConn(ctx context.Context, conn net.Conn) {
s.eventLoop(client)
}
// sendServerIdent sends the initial server identification packet to the Client.
// eventLoop reads packets from the session and dispatches handlers.
// SenderWorker must already be running before eventLoop is entered.
func (s *Server) eventLoop(client *session.Session) {
defer client.Cancel()
for {
if !client.Scanner.Scan() {
return
}
// Reference the next packet
packet := client.Scanner.Bytes()
packet = append(packet, '\r', '\n') // Re-append delimiter
// Verify packet and obtain type
packetType, ok := verifyPacket(packet, client)
if !ok {
continue
}
// Run handler
handler := s.getHandler(packetType)
handler(client, packet)
}
}
// sendServerIdent sends the initial server identification packet to the client.
// It returns an error if writing to the connection fails.
func sendServerIdent(conn io.Writer) (err error) {
packet := protocol.ServerIdent{
@@ -98,31 +132,17 @@ func sendServerIdent(conn io.Writer) (err error) {
return
}
// loginData holds the data extracted from the Client's login packets.
type loginData struct {
clientChallenge string // Optional Client challenge for authentication
callsign string // Callsign of the Client
cid int // Cert ID
realName string // Real name
networkRating NetworkRating // Network rating of the Client
maxNetworkRating NetworkRating // Maximum allowed network rating (what is stored in the database)
protoRevision int // Protocol revision
loginTime time.Time // Time of login
clientId uint16 // Client ID
isAtc bool // True if the Client is an ATC, false if a pilot
}
// ErrInvalidAddPacket is returned when the add packet from the Client is invalid.
// ErrInvalidAddPacket is returned when the add packet from the client is invalid.
var ErrInvalidAddPacket = errors.New("invalid add packet")
// ErrInvalidIDPacket is returned when the ID packet from the Client is invalid.
// ErrInvalidIDPacket is returned when the ID packet from the client is invalid.
var ErrInvalidIDPacket = errors.New("invalid ID packet")
// readLoginPackets reads the two expected login packets from the Client:
// the Client identification packet and the add packet.
// It parses these packets to extract the Client's data and returns it in a loginData struct.
// If any errors occur during reading or parsing, it sends an error to the Client and returns an error.
func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, token string, err error) {
// readLoginPackets reads the two expected login packets from the client:
// the client identification packet and the add packet.
// It parses these packets to extract the client's data and returns it in a LoginData struct.
// If any errors occur during reading or parsing, it sends an error to the client and returns an error.
func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data session.LoginData, token string, err error) {
// Client ident
if !scanner.Scan() {
err = ErrInvalidIDPacket
@@ -139,10 +159,10 @@ func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, to
}
addPacket := append([]byte{}, scanner.Bytes()...)
// Check if the Client sent a challenge field
// Check if the client sent a challenge field
if countFields(idPacket) == 9 {
// Extract the challenge
data.clientChallenge = string(getField(idPacket, 8))
data.ClientChallenge = string(getField(idPacket, 8))
// Extract the client ID
var clientId uint64
@@ -152,7 +172,7 @@ func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, to
sendError(conn, SyntaxError, "Error parsing client ID")
return
}
data.clientId = uint16(clientId)
data.ClientID = uint16(clientId)
}
if len(addPacket) < 16 {
@@ -161,11 +181,11 @@ func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, to
return
}
// Determine Client type
// Determine client type
var prefix string
switch string(addPacket[:3]) {
case "#AA":
data.isAtc = true
data.IsAtc = true
prefix = "#AA"
case "#AP":
prefix = "#AP"
@@ -175,7 +195,7 @@ func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, to
return
}
if data.isAtc {
if data.IsAtc {
if countFields(addPacket) != 7 {
err = ErrInvalidAddPacket
sendError(conn, SyntaxError, "Invalid number of fields in ATC add packet")
@@ -190,16 +210,16 @@ func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, to
}
if callsign, found := bytes.CutPrefix(getField(addPacket, 0), []byte(prefix)); found {
data.callsign = string(callsign)
data.Callsign = string(callsign)
} else {
sendError(conn, SyntaxError, "Invalid callsign in add packet")
err = ErrInvalidAddPacket
return
}
if data.isAtc {
data.realName = string(getField(addPacket, 2))
if data.cid, err = strconv.Atoi(string(getField(addPacket, 3))); err != nil {
if data.IsAtc {
data.RealName = string(getField(addPacket, 2))
if data.CID, err = strconv.Atoi(string(getField(addPacket, 3))); err != nil {
err = ErrInvalidAddPacket
sendError(conn, SyntaxError, "Invalid CID in ATC add packet")
return
@@ -211,14 +231,14 @@ func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, to
sendError(conn, SyntaxError, "Invalid network rating in pilot add packet")
return
}
data.networkRating = NetworkRating(networkRating)
if data.protoRevision, err = strconv.Atoi(string(getField(addPacket, 6))); err != nil {
data.NetworkRating = NetworkRating(networkRating)
if data.ProtoRevision, err = strconv.Atoi(string(getField(addPacket, 6))); err != nil {
err = ErrInvalidAddPacket
sendError(conn, SyntaxError, "Invalid protocol revision in ATC add packet")
return
}
} else {
if data.cid, err = strconv.Atoi(string(getField(addPacket, 2))); err != nil {
if data.CID, err = strconv.Atoi(string(getField(addPacket, 2))); err != nil {
err = ErrInvalidAddPacket
sendError(conn, SyntaxError, "Invalid CID in pilot add packet")
return
@@ -230,35 +250,35 @@ func readLoginPackets(conn net.Conn, scanner *bufio.Scanner) (data loginData, to
sendError(conn, SyntaxError, "Invalid network rating in pilot add packet")
return
}
data.networkRating = NetworkRating(networkRating)
if data.protoRevision, err = strconv.Atoi(string(getField(addPacket, 5))); err != nil {
data.NetworkRating = NetworkRating(networkRating)
if data.ProtoRevision, err = strconv.Atoi(string(getField(addPacket, 5))); err != nil {
err = ErrInvalidAddPacket
sendError(conn, SyntaxError, "Invalid protocol revision in pilot add packet")
return
}
data.realName = string(getField(addPacket, 7))
data.RealName = string(getField(addPacket, 7))
}
if data.protoRevision < 100 || data.protoRevision > 101 {
if data.ProtoRevision < 100 || data.ProtoRevision > 101 {
err = ErrInvalidAddPacket
sendError(conn, InvalidProtocolRevisionError, "Invalid protocol revision")
return
}
data.loginTime = time.Now()
data.LoginTime = time.Now()
return
}
func (s *Server) attemptAuthentication(client *Client, token string) (err error) {
func (s *Server) attemptAuthentication(client *session.Session, token string) (err error) {
// Check vatsim auth compatibility
if client.clientChallenge != "" {
if err = client.authState.Initialize(
client.clientId,
[]byte(client.clientChallenge),
if client.ClientChallenge != "" {
if err = client.Auth.Initialize(
client.ClientID,
[]byte(client.ClientChallenge),
); err != nil {
err = ErrInvalidIDPacket
sendError(client.conn, UnauthorizedSoftwareError, "Client incompatible with auth challenges")
sendError(client.Conn, UnauthorizedSoftwareError, "Client incompatible with auth challenges")
return
}
}
@@ -275,7 +295,7 @@ func (s *Server) attemptAuthentication(client *Client, token string) (err error)
var jwtToken *auth.JwtToken
if jwtToken, err = auth.ParseJwtToken(token, []byte(jwtSecret)); err != nil {
err = ErrInvalidAddPacket
sendError(client.conn, InvalidLogonError, invalidLogonMsg)
sendError(client.Conn, InvalidLogonError, invalidLogonMsg)
return
}
@@ -283,26 +303,26 @@ func (s *Server) attemptAuthentication(client *Client, token string) (err error)
if claims.TokenType != "fsd" {
err = ErrInvalidAddPacket
sendError(client.conn, InvalidLogonError, invalidLogonMsg)
sendError(client.Conn, InvalidLogonError, invalidLogonMsg)
return
}
if client.cid != claims.CID {
if client.CID != claims.CID {
err = ErrInvalidAddPacket
sendError(client.conn, RequestedLevelTooHighError, invalidLogonMsg)
sendError(client.Conn, RequestedLevelTooHighError, invalidLogonMsg)
return
}
if client.networkRating > claims.NetworkRating {
if client.NetworkRating > claims.NetworkRating {
err = ErrInvalidAddPacket
sendError(client.conn, RequestedLevelTooHighError, "Requested level too high")
sendError(client.Conn, RequestedLevelTooHighError, "Requested level too high")
return
}
if client.networkRating < NetworkRatingObserver {
if client.NetworkRating < NetworkRatingObserver {
err = ErrInvalidAddPacket
sendError(client.conn, CertificateSuspendedError, "Certificate inactive or suspended")
sendError(client.Conn, CertificateSuspendedError, "Certificate inactive or suspended")
return
}
client.maxNetworkRating = claims.NetworkRating
client.MaxNetworkRating = claims.NetworkRating
return
}
@@ -311,76 +331,76 @@ func (s *Server) attemptAuthentication(client *Client, token string) (err error)
password := token
// Attempt to fetch user
user, err := s.dbRepo.UserRepo.GetUserByCID(client.cid)
user, err := s.dbRepo.UserRepo.GetUserByCID(client.CID)
if err != nil {
err = ErrInvalidAddPacket
sendError(client.conn, InvalidLogonError, invalidLogonMsg)
sendError(client.Conn, InvalidLogonError, invalidLogonMsg)
return
}
// Verify password hash
if !s.dbRepo.UserRepo.VerifyPasswordHash(password, user.Password) {
err = ErrInvalidAddPacket
sendError(client.conn, InvalidLogonError, invalidLogonMsg)
sendError(client.Conn, InvalidLogonError, invalidLogonMsg)
return
}
// Verify network rating
if client.networkRating > NetworkRating(user.NetworkRating) {
if client.NetworkRating > NetworkRating(user.NetworkRating) {
err = ErrInvalidAddPacket
sendError(client.conn, RequestedLevelTooHighError, "Requested level too high")
sendError(client.Conn, RequestedLevelTooHighError, "Requested level too high")
return
}
if client.networkRating < NetworkRatingObserver {
if client.NetworkRating < NetworkRatingObserver {
err = ErrInvalidAddPacket
sendError(client.conn, CertificateSuspendedError, "Certificate inactive or suspended")
sendError(client.Conn, CertificateSuspendedError, "Certificate inactive or suspended")
return
}
client.maxNetworkRating = NetworkRating(user.NetworkRating)
client.MaxNetworkRating = NetworkRating(user.NetworkRating)
return
}
func (s *Server) broadcastAddPacket(client *Client) {
func (s *Server) broadcastAddPacket(client *session.Session) {
var packet string
if client.isAtc {
if client.IsAtc {
packet = fmt.Sprintf(
"#AA%s:SERVER:%s:%d::%d:%d\r\n",
client.callsign,
client.realName,
client.cid,
client.networkRating,
client.protoRevision)
client.Callsign,
client.RealName,
client.CID,
client.NetworkRating,
client.ProtoRevision)
} else {
packet = fmt.Sprintf(
"#AP%s:SERVER:%d::%d:%d:1:%s\r\n",
client.callsign,
client.cid,
client.networkRating,
client.protoRevision,
client.realName)
client.Callsign,
client.CID,
client.NetworkRating,
client.ProtoRevision,
client.RealName)
}
broadcastAll(s.postOffice, client, []byte(packet))
}
func (s *Server) broadcastDisconnectPacket(client *Client) {
func (s *Server) broadcastDisconnectPacket(client *session.Session) {
packet := strings.Builder{}
if client.isAtc {
if client.IsAtc {
packet.WriteString("#DA")
} else {
packet.WriteString("#DP")
}
packet.WriteString(client.callsign)
packet.WriteString(client.Callsign)
packet.WriteString(":SERVER:")
packet.WriteString(strconv.Itoa(client.cid))
packet.WriteString(strconv.Itoa(client.CID))
packet.WriteString("\r\n")
broadcastAll(s.postOffice, client, []byte(packet.String()))
}
func (s *Server) sendMotd(client *Client) (err error) {
func (s *Server) sendMotd(client *session.Session) (err error) {
welcomeMsg := db.GetWelcomeMessage(s.dbRepo.ConfigRepo)
if welcomeMsg != "" {
lines := strings.Split(welcomeMsg, "\n")
@@ -397,16 +417,15 @@ func (s *Server) sendMotd(client *Client) (err error) {
return
}
// sendServerTextMessage synchronously sends a server #TM to the client's socket
func (s *Server) sendServerTextMessage(client *Client, msg string) (err error) {
// sendServerTextMessage enqueues a server #TM via the session send channel.
func (s *Server) sendServerTextMessage(client *session.Session, msg string) (err error) {
packet := strings.Builder{}
packet.Grow(32 + len(msg))
packet.WriteString("#TMserver:")
packet.WriteString(client.callsign)
packet.WriteString(client.Callsign)
packet.WriteByte(':')
packet.WriteString(msg)
packet.WriteString("\r\n")
_, err = client.conn.Write([]byte(packet.String()))
return
return client.Send(packet.String())
}

172
fsd/handler.go
View File

@@ -7,6 +7,8 @@ import (
"strconv"
"strings"
"time"
"github.com/renorris/openfsd/internal/session"
)
func (s *Server) getHandler(packetType PacketType) handlerFunc {
@@ -44,17 +46,17 @@ func (s *Server) getHandler(packetType PacketType) handlerFunc {
}
}
func (s *Server) emptyHandler(client *Client, packet []byte) {
func (s *Server) emptyHandler(client *session.Session, packet []byte) {
slog.Error("empty handler called")
return
}
func (s *Server) handleTextMessage(client *Client, packet []byte) {
func (s *Server) handleTextMessage(client *session.Session, packet []byte) {
recipient := getField(packet, 1)
// ATC chat
if string(recipient) == "@49999" {
if !client.isAtc {
if !client.IsAtc {
return
}
broadcastRangedAtcOnly(s.postOffice, client, packet)
@@ -75,7 +77,7 @@ func (s *Server) handleTextMessage(client *Client, packet []byte) {
// Server-wide broadcast message
if string(recipient) == "*" {
if client.networkRating < NetworkRatingSupervisor {
if client.NetworkRating < NetworkRatingSupervisor {
return
}
broadcastAll(s.postOffice, client, packet)
@@ -96,31 +98,31 @@ func (s *Server) handleTextMessage(client *Client, packet []byte) {
sendDirectOrErr(s.postOffice, client, recipient, packet)
}
func (s *Server) handleATCPosition(client *Client, packet []byte) {
func (s *Server) handleATCPosition(client *session.Session, packet []byte) {
// Verify and set facility type
facilityType, err := strconv.ParseInt(string(getField(packet, 2)), 10, 32)
if err != nil {
client.sendError(SyntaxError, "Invalid facility type")
client.SendError(SyntaxError, "Invalid facility type")
return
}
if !isAllowedFacilityType(client.networkRating, int(facilityType)) {
client.sendError(InvalidPositionForRatingError, "Invalid position for rating")
client.cancelCtx()
if !isAllowedFacilityType(client.NetworkRating, int(facilityType)) {
client.SendError(InvalidPositionForRatingError, "Invalid position for rating")
client.Cancel()
return
}
client.facilityType = int(facilityType)
client.FacilityType = int(facilityType)
// Extract location and visibility range
lat, lon, ok := parseLatLon(packet, 5, 6)
if !ok {
client.sendError(SyntaxError, "Invalid latitude/longitude")
client.SendError(SyntaxError, "Invalid latitude/longitude")
return
}
visRange, ok := parseVisRange(packet, 3)
if !ok {
client.sendError(SyntaxError, "Invalid visibility range")
client.SendError(SyntaxError, "Invalid visibility range")
return
}
@@ -130,14 +132,14 @@ func (s *Server) handleATCPosition(client *Client, packet []byte) {
// Broadcast position update
broadcastRanged(s.postOffice, client, packet)
client.lastUpdated.Store(time.Now())
client.LastUpdated.Store(time.Now())
}
// handlePilotPosition handles logic for 0.2hz `@` pilot position updates
func (s *Server) handlePilotPosition(client *Client, packet []byte) {
func (s *Server) handlePilotPosition(client *session.Session, packet []byte) {
lat, lon, ok := parseLatLon(packet, 4, 5)
if !ok {
client.sendError(SyntaxError, "Invalid latitude/longitude")
client.SendError(SyntaxError, "Invalid latitude/longitude")
return
}
@@ -150,30 +152,30 @@ func (s *Server) handlePilotPosition(client *Client, packet []byte) {
broadcastRanged(s.postOffice, client, packet)
// Update state
client.transponder.Store(string(getField(packet, 2)))
client.Transponder.Store(string(getField(packet, 2)))
groundspeed, _ := strconv.Atoi(string(getField(packet, 7)))
client.groundspeed.Store(int32(groundspeed))
client.Groundspeed.Store(int32(groundspeed))
altitude, _ := strconv.Atoi(string(getField(packet, 6)))
client.altitude.Store(int32(altitude))
client.Altitude.Store(int32(altitude))
pbhUint, _ := strconv.ParseUint(string(getField(packet, 8)), 10, 32)
_, _, heading := pitchBankHeading(uint32(pbhUint))
client.heading.Store(int32(heading))
client.Heading.Store(int32(heading))
client.lastUpdated.Store(time.Now())
client.LastUpdated.Store(time.Now())
// Check if we need to update the sendfast state
if client.protoRevision == 101 {
if client.sendFastEnabled {
if (client.closestVelocityClientDistance / 1852.0) > 5.0 { // 5.0 nautical miles
client.sendFastEnabled = false
if client.ProtoRevision == 101 {
if client.SendFastEnabled {
if (client.ClosestVelocityClientDistance / 1852.0) > 5.0 { // 5.0 nautical miles
client.SendFastEnabled = false
sendDisableSendFastPacket(client)
}
} else {
if (client.closestVelocityClientDistance / 1852.0) < 5.0 { // 5.0 nautical miles
client.sendFastEnabled = true
if (client.ClosestVelocityClientDistance / 1852.0) < 5.0 { // 5.0 nautical miles
client.SendFastEnabled = true
sendEnableSendFastPacket(client)
}
}
@@ -181,37 +183,37 @@ func (s *Server) handlePilotPosition(client *Client, packet []byte) {
}
// handleFastPilotPosition handles logic for fast `^`, stopped `#ST`, and slow `#SL` pilot position updates
func (s *Server) handleFastPilotPosition(client *Client, packet []byte) {
func (s *Server) handleFastPilotPosition(client *session.Session, packet []byte) {
// Broadcast position update
broadcastRangedVelocity(s.postOffice, client, packet)
}
// handleDelete handles logic for Delete ATC `#DA` and Delete Pilot `#DP` packets
func (s *Server) handleDelete(client *Client, packet []byte) {
func (s *Server) handleDelete(client *session.Session, packet []byte) {
// Broadcast delete packet
broadcastAll(s.postOffice, client, packet)
// Cancel context. Writer worker will close the connection
client.cancelCtx()
client.Cancel()
}
// handleSquawkbox handles logic for Squawkbox `#SB` packets
func (s *Server) handleSquawkbox(client *Client, packet []byte) {
func (s *Server) handleSquawkbox(client *session.Session, packet []byte) {
// Forward packet to recipient
recipient := getField(packet, 1)
sendDirectOrErr(s.postOffice, client, recipient, packet)
}
// handleProcontroller handles logic for Pro Controller `#PC` packets
func (s *Server) handleProcontroller(client *Client, packet []byte) {
func (s *Server) handleProcontroller(client *session.Session, packet []byte) {
// ATC-only packet
if !client.isAtc {
if !client.IsAtc {
return
}
recipient := getField(packet, 1)
if len(recipient) < 2 {
client.sendError(SyntaxError, "Invalid recipient")
client.SendError(SyntaxError, "Invalid recipient")
return
}
pcType := getField(packet, 3)
@@ -244,8 +246,8 @@ func (s *Server) handleProcontroller(client *Client, packet []byte) {
"ST": // Set flight strip
// Only active ATC above OBS
if client.facilityType <= 0 {
client.sendError(InvalidControlError, "Invalid control")
if client.FacilityType <= 0 {
client.SendError(InvalidControlError, "Invalid control")
return
}
if recipient[0] == '@' {
@@ -256,7 +258,7 @@ func (s *Server) handleProcontroller(client *Client, packet []byte) {
}
}
func (s *Server) handleClientQuery(client *Client, packet []byte) {
func (s *Server) handleClientQuery(client *session.Session, packet []byte) {
recipient := getField(packet, 1)
queryType := getField(packet, 2)
@@ -286,8 +288,8 @@ func (s *Server) handleClientQuery(client *Client, packet []byte) {
"NEWINFO": // Broadcast new ATIS info
// ATC only
if !client.isAtc {
client.sendError(InvalidControlError, "Invalid control")
if !client.IsAtc {
client.SendError(InvalidControlError, "Invalid control")
return
}
forwardClientQuery(s.postOffice, client, packet)
@@ -307,8 +309,8 @@ func (s *Server) handleClientQuery(client *Client, packet []byte) {
"IPC": // Force squawk code change
// ATC above OBS facility only
if !client.isAtc || client.facilityType <= 0 {
client.sendError(InvalidControlError, "Invalid control")
if !client.IsAtc || client.FacilityType <= 0 {
client.SendError(InvalidControlError, "Invalid control")
return
}
forwardClientQuery(s.postOffice, client, packet)
@@ -326,81 +328,81 @@ func (s *Server) handleClientQuery(client *Client, packet []byte) {
}
// Require >= SUP for interrogations
if client.networkRating < NetworkRatingSupervisor {
client.sendError(InvalidControlError, "Invalid control")
if client.NetworkRating < NetworkRatingSupervisor {
client.SendError(InvalidControlError, "Invalid control")
return
}
forwardClientQuery(s.postOffice, client, packet)
}
}
func (s *Server) handleClientQueryATCRequest(client *Client, packet []byte) {
func (s *Server) handleClientQueryATCRequest(client *session.Session, packet []byte) {
if countFields(packet) != 4 {
client.sendError(SyntaxError, "Invalid ATC request")
client.SendError(SyntaxError, "Invalid ATC request")
return
}
targetCallsign := getField(packet, 3)
targetClient, err := s.postOffice.find(string(targetCallsign))
if err != nil {
client.sendError(NoSuchCallsignError, "No such callsign")
client.SendError(NoSuchCallsignError, "No such callsign")
return
}
var p string
if targetClient.facilityType > 0 {
p = fmt.Sprintf("$CRSERVER:%s:ATC:Y:%s\r\n", client.callsign, targetCallsign)
if targetClient.FacilityType > 0 {
p = fmt.Sprintf("$CRSERVER:%s:ATC:Y:%s\r\n", client.Callsign, targetCallsign)
} else {
p = fmt.Sprintf("$CRSERVER:%s:ATC:N:%s\r\n", client.callsign, targetCallsign)
p = fmt.Sprintf("$CRSERVER:%s:ATC:N:%s\r\n", client.Callsign, targetCallsign)
}
client.send(p)
client.Send(p)
}
func (s *Server) handleClientQueryIPRequest(client *Client, packet []byte) {
ip := strings.SplitN(client.conn.RemoteAddr().String(), ":", 2)[0]
p := fmt.Sprintf("$CRSERVER:%s:IP:%s\r\n", client.callsign, ip)
client.send(p)
func (s *Server) handleClientQueryIPRequest(client *session.Session, packet []byte) {
ip := strings.SplitN(client.Conn.RemoteAddr().String(), ":", 2)[0]
p := fmt.Sprintf("$CRSERVER:%s:IP:%s\r\n", client.Callsign, ip)
client.Send(p)
}
func (s *Server) handleClientQueryFlightplanRequest(client *Client, packet []byte) {
if !client.isAtc {
func (s *Server) handleClientQueryFlightplanRequest(client *session.Session, packet []byte) {
if !client.IsAtc {
return
}
if countFields(packet) != 4 {
client.sendError(SyntaxError, "Invalid flightplan request syntax")
client.SendError(SyntaxError, "Invalid flightplan request syntax")
return
}
targetCallsign := string(getField(packet, 3))
targetClient, err := s.postOffice.find(targetCallsign)
if err != nil {
client.sendError(NoSuchCallsignError, "No such callsign: "+targetCallsign)
client.SendError(NoSuchCallsignError, "No such callsign: "+targetCallsign)
return
}
fplInfo := targetClient.flightPlan.Load()
fplInfo := targetClient.FlightPlan.Load()
if fplInfo == "" {
return
}
beaconCode := targetClient.assignedBeaconCode.Load()
beaconCode := targetClient.AssignedBeaconCode.Load()
if beaconCode == "" {
beaconCode = "0"
}
// Send flightplan packet
fplPacket := buildFileFlightplanPacket(targetCallsign, "*A", fplInfo)
client.send(fplPacket)
client.Send(fplPacket)
// Send assigned beacon code
bcPacket := buildBeaconCodePacket("server", client.callsign, targetCallsign, beaconCode)
client.send(bcPacket)
bcPacket := buildBeaconCodePacket("server", client.Callsign, targetCallsign, beaconCode)
client.Send(bcPacket)
// TODO: research any other data that should be sent here
}
func (s *Server) handleMetarRequest(client *Client, packet []byte) {
func (s *Server) handleMetarRequest(client *session.Session, packet []byte) {
recipient := getField(packet, 1)
staticField := getField(packet, 2)
icaoCode := getField(packet, 3)
@@ -409,11 +411,11 @@ func (s *Server) handleMetarRequest(client *Client, packet []byte) {
return
}
s.metarService.fetchAndSendMetar(client.ctx, client, string(icaoCode))
s.metarService.fetchAndSendMetar(client.Ctx, client, string(icaoCode))
}
func (s *Server) handleKillRequest(client *Client, packet []byte) {
if client.networkRating < NetworkRatingSupervisor {
func (s *Server) handleKillRequest(client *session.Session, packet []byte) {
if client.NetworkRating < NetworkRatingSupervisor {
return
}
@@ -421,37 +423,37 @@ func (s *Server) handleKillRequest(client *Client, packet []byte) {
recipient := getField(packet, 1)
victim, err := s.postOffice.find(string(recipient))
if err != nil {
client.sendError(NoSuchCallsignError, "No such callsign")
client.SendError(NoSuchCallsignError, "No such callsign")
return
}
// Closing the context of the victim client will eventually cause it to disconnect
victim.cancelCtx()
victim.Cancel()
}
func (s *Server) handleAuthChallenge(client *Client, packet []byte) {
if client.clientChallenge == "" {
client.sendError(UnauthorizedSoftwareError, "Cannot reply to auth challenge since no initial challenge was recieved")
func (s *Server) handleAuthChallenge(client *session.Session, packet []byte) {
if client.ClientChallenge == "" {
client.SendError(UnauthorizedSoftwareError, "Cannot reply to auth challenge since no initial challenge was recieved")
return
}
challenge := getField(packet, 2)
resp := client.authState.GetResponseForChallenge(challenge)
client.authState.UpdateState(&resp)
resp := client.Auth.GetResponseForChallenge(challenge)
client.Auth.UpdateState(&resp)
respPacket := strings.Builder{}
respPacket.WriteString("$ZRSERVER:")
respPacket.WriteString(client.callsign)
respPacket.WriteString(client.Callsign)
respPacket.WriteByte(':')
respPacket.Write(resp[:])
respPacket.WriteString("\r\n")
client.send(respPacket.String())
client.Send(respPacket.String())
}
func (s *Server) handleHandoff(client *Client, packet []byte) {
func (s *Server) handleHandoff(client *session.Session, packet []byte) {
// Active >OBS ATC only
if !client.isAtc || client.facilityType <= 1 {
if !client.IsAtc || client.FacilityType <= 1 {
return
}
@@ -459,16 +461,16 @@ func (s *Server) handleHandoff(client *Client, packet []byte) {
sendDirectOrErr(s.postOffice, client, recipient, packet)
}
func (s *Server) handleFileFlightplan(client *Client, packet []byte) {
func (s *Server) handleFileFlightplan(client *session.Session, packet []byte) {
fplInfo := extractFlightplanInfoSection(packet)
client.flightPlan.Store(fplInfo)
client.FlightPlan.Store(fplInfo)
broadcastPacket := buildFileFlightplanPacket(client.callsign, "*A", fplInfo)
broadcastPacket := buildFileFlightplanPacket(client.Callsign, "*A", fplInfo)
broadcastAllATC(s.postOffice, client, []byte(broadcastPacket))
}
func (s *Server) handleAmendFlightplan(client *Client, packet []byte) {
if !client.isAtc || client.facilityType <= 0 {
func (s *Server) handleAmendFlightplan(client *session.Session, packet []byte) {
if !client.IsAtc || client.FacilityType <= 0 {
return
}
@@ -477,11 +479,11 @@ func (s *Server) handleAmendFlightplan(client *Client, packet []byte) {
targetCallsign := string(getField(packet, 2))
targetClient, err := s.postOffice.find(targetCallsign)
if err != nil {
client.sendError(NoSuchCallsignError, "No such callsign: "+targetCallsign)
client.SendError(NoSuchCallsignError, "No such callsign: "+targetCallsign)
return
}
targetClient.flightPlan.Store(fplInfo)
targetClient.FlightPlan.Store(fplInfo)
broadcastPacket := buildAmendFlightplanPacket(client.callsign, "*A", targetCallsign, fplInfo)
broadcastPacket := buildAmendFlightplanPacket(client.Callsign, "*A", targetCallsign, fplInfo)
broadcastAllATC(s.postOffice, client, []byte(broadcastPacket))
}

39
fsd/http_service.go
View File

@@ -12,7 +12,8 @@ import (
"github.com/gin-gonic/gin"
"github.com/renorris/openfsd/db"
"github.com/renorris/openfsd/internal/auth"
"github.com/renorris/openfsd/internal/auth"
"github.com/renorris/openfsd/internal/session"
)
// runServiceHTTP starts the admin service HTTP server used for
@@ -101,7 +102,7 @@ func (s *Server) handleGetOnlineUsers(c *gin.Context) {
mapLen := len(s.postOffice.clientMap)
s.postOffice.clientMapLock.RUnlock()
clientMap := make(map[string]*Client, mapLen+16)
clientMap := make(map[string]*session.Session, mapLen+16)
s.postOffice.clientMapLock.RLock()
maps.Copy(clientMap, s.postOffice.clientMap)
@@ -113,34 +114,34 @@ func (s *Server) handleGetOnlineUsers(c *gin.Context) {
}
for _, client := range clientMap {
latLon := client.latLon()
latLon := client.LatLon()
genData := OnlineUserGeneralData{
Callsign: client.callsign,
CID: client.cid,
Name: client.realName,
NetworkRating: int(client.networkRating),
MaxNetworkRating: int(client.maxNetworkRating),
Callsign: client.Callsign,
CID: client.CID,
Name: client.RealName,
NetworkRating: int(client.NetworkRating),
MaxNetworkRating: int(client.MaxNetworkRating),
Latitude: latLon[0],
Longitude: latLon[1],
LogonTime: client.loginTime,
LastUpdated: client.lastUpdated.Load(),
LogonTime: client.LoginTime,
LastUpdated: client.LastUpdated.Load(),
}
if client.isAtc {
if client.IsAtc {
atc := OnlineUserATC{
OnlineUserGeneralData: genData,
Frequency: client.frequency.Load(),
Facility: client.facilityType,
VisRange: int(client.visRange.Load() * 0.000539957), // Convert meters to nautical miles
Frequency: client.Frequency.Load(),
Facility: client.FacilityType,
VisRange: int(client.VisRange.Load() * 0.000539957), // Convert meters to nautical miles
}
resData.ATC = append(resData.ATC, atc)
} else {
pilot := OnlineUserPilot{
OnlineUserGeneralData: genData,
Altitude: int(client.altitude.Load()),
Groundspeed: int(client.groundspeed.Load()),
Heading: int(client.heading.Load()),
Transponder: client.transponder.Load(),
Altitude: int(client.Altitude.Load()),
Groundspeed: int(client.Groundspeed.Load()),
Heading: int(client.Heading.Load()),
Transponder: client.Transponder.Load(),
}
resData.Pilots = append(resData.Pilots, pilot)
}
@@ -172,7 +173,7 @@ func (s *Server) handleKickUser(c *gin.Context) {
}
// Cancelling the context will cause the client's event loop to close
client.cancelCtx()
client.Cancel()
c.AbortWithStatus(http.StatusNoContent)
}

16
fsd/metar.go
View File

@@ -3,10 +3,12 @@ package fsd
import (
"bytes"
"context"
"fmt"
"io"
"log/slog"
"net/http"
"strings"
"github.com/renorris/openfsd/internal/session"
)
type metarService struct {
@@ -16,7 +18,7 @@ type metarService struct {
}
type metarRequest struct {
client *Client
client *session.Session
icaoCode string
}
@@ -67,7 +69,7 @@ func (s *metarService) handleMetarRequest(req *metarRequest) {
resBody := buf.Bytes()
if bytes.Count(resBody, []byte("\n")) != 2 {
fmt.Println("NOAA METAR response was invalid")
slog.Debug("NOAA METAR response was invalid")
sendMetarServiceError(req)
return
}
@@ -78,8 +80,8 @@ func (s *metarService) handleMetarRequest(req *metarRequest) {
// Second line is METAR and ends with \n
resBody = resBody[:bytes.IndexByte(resBody, '\n')+1]
packet := buildMetarResponsePacket(req.client.callsign, resBody)
req.client.send(packet)
packet := buildMetarResponsePacket(req.client.Callsign, resBody)
req.client.Send(packet)
}
func buildMetarResponsePacket(callsign string, metar []byte) string {
@@ -101,7 +103,7 @@ func buildMetarRequestURL(icaoCode string) string {
}
func sendMetarServiceError(req *metarRequest) {
req.client.sendError(NoWeatherProfileError, metarServiceErrString(req.icaoCode))
req.client.SendError(NoWeatherProfileError, metarServiceErrString(req.icaoCode))
}
func metarServiceErrString(icaoCode string) string {
@@ -114,7 +116,7 @@ func metarServiceErrString(icaoCode string) string {
// fetchAndSendMetar fetches a METAR observation for a given ICAO code and sends it to the client once received.
// This function returns immediately once the request has been queued.
func (s *metarService) fetchAndSendMetar(ctx context.Context, client *Client, icaoCode string) {
func (s *metarService) fetchAndSendMetar(ctx context.Context, client *session.Session, icaoCode string) {
select {
case <-ctx.Done():
case s.metarRequests <- metarRequest{client: client, icaoCode: icaoCode}:

79
fsd/metar_test.go
View File

@@ -8,45 +8,24 @@ import (
"net/http"
"strings"
"testing"
"github.com/renorris/openfsd/internal/session"
)
// mockClient simulates a Client for capturing sent packets.
type mockClient struct {
*Client
sentPackets []string
// newMockSession creates a session with a buffered outbound channel for tests.
func newMockSession(callsign string) *session.Session {
return session.New(context.Background(), nil, nil, session.LoginData{Callsign: callsign})
}
// newMockClient creates a mockClient with a valid sendChan and ctx.
func newMockClient(callsign string) *mockClient {
ctx, cancel := context.WithCancel(context.Background())
client := &Client{
ctx: ctx,
cancelCtx: cancel,
sendChan: make(chan string, 32), // Buffered to prevent blocking
loginData: loginData{callsign: callsign},
}
return &mockClient{
Client: client,
sentPackets: []string{},
}
}
// send overrides Client's send method to capture packets.
func (c *mockClient) send(packet string) error {
c.sentPackets = append(c.sentPackets, packet)
return nil
}
// collectPackets drains the sendChan and returns all sent packets.
func (c *mockClient) collectPackets() []string {
packets := append([]string{}, c.sentPackets...)
// collectPackets drains the outbound send channel and returns all sent packets.
func collectPackets(s *session.Session) []string {
var packets []string
for {
select {
case packet := <-c.sendChan:
packets = append(packets, packet)
default:
pkt, ok := s.DequeueOutbound()
if !ok {
return packets
}
packets = append(packets, pkt)
}
}
@@ -126,14 +105,14 @@ func TestBuildMetarResponsePacket(t *testing.T) {
// TestSendMetarServiceError verifies that sendMetarServiceError sends the correct error packet to the client.
func TestSendMetarServiceError(t *testing.T) {
mockClient := newMockClient("TEST")
mock := newMockSession("TEST")
req := &metarRequest{
client: mockClient.Client,
client: mock,
icaoCode: "KJFK",
}
sendMetarServiceError(req)
packets := mockClient.collectPackets()
packets := collectPackets(mock)
expectedPacket := "$ERserver:unknown:9::Error fetching METAR for KJFK\r\n"
if len(packets) != 1 {
t.Errorf("expected 1 packet sent, got %d", len(packets))
@@ -155,15 +134,15 @@ func TestHandleMetarRequest_Success(t *testing.T) {
httpClient: &http.Client{Transport: mockTransport},
}
mockClient := newMockClient("TEST")
mock := newMockSession("TEST")
req := &metarRequest{
client: mockClient.Client,
client: mock,
icaoCode: "KJFK",
}
service.handleMetarRequest(req)
packets := mockClient.collectPackets()
packets := collectPackets(mock)
if len(packets) != 1 {
t.Errorf("expected 1 packet sent, got %d", len(packets))
}
@@ -185,15 +164,15 @@ func TestHandleMetarRequest_HTTPError(t *testing.T) {
httpClient: &http.Client{Transport: mockTransport},
}
mockClient := newMockClient("TEST")
mock := newMockSession("TEST")
req := &metarRequest{
client: mockClient.Client,
client: mock,
icaoCode: "INVALID",
}
service.handleMetarRequest(req)
packets := mockClient.collectPackets()
packets := collectPackets(mock)
expectedPacket := "$ERserver:unknown:9::Error fetching METAR for INVALID\r\n"
if len(packets) != 1 {
t.Errorf("expected 1 packet sent, got %d", len(packets))
@@ -210,15 +189,15 @@ func TestHandleMetarRequest_NetworkError(t *testing.T) {
httpClient: &http.Client{Transport: mockTransport},
}
mockClient := newMockClient("TEST")
mock := newMockSession("TEST")
req := &metarRequest{
client: mockClient.Client,
client: mock,
icaoCode: "KJFK",
}
service.handleMetarRequest(req)
packets := mockClient.collectPackets()
packets := collectPackets(mock)
expectedPacket := "$ERserver:unknown:9::Error fetching METAR for KJFK\r\n"
if len(packets) != 1 {
t.Errorf("expected 1 packet sent, got %d", len(packets))
@@ -240,15 +219,15 @@ func TestHandleMetarRequest_InvalidResponse(t *testing.T) {
httpClient: &http.Client{Transport: mockTransport},
}
mockClient := newMockClient("TEST")
mock := newMockSession("TEST")
req := &metarRequest{
client: mockClient.Client,
client: mock,
icaoCode: "KJFK",
}
service.handleMetarRequest(req)
packets := mockClient.collectPackets()
packets := collectPackets(mock)
expectedPacket := "$ERserver:unknown:9::Error fetching METAR for KJFK\r\n"
if len(packets) != 1 {
t.Errorf("expected 1 packet sent, got %d", len(packets))
@@ -270,15 +249,15 @@ func TestHandleMetarRequest_MoreThanTwoLines(t *testing.T) {
httpClient: &http.Client{Transport: mockTransport},
}
mockClient := newMockClient("TEST")
mock := newMockSession("TEST")
req := &metarRequest{
client: mockClient.Client,
client: mock,
icaoCode: "KJFK",
}
service.handleMetarRequest(req)
packets := mockClient.collectPackets()
packets := collectPackets(mock)
expectedPacket := "$ERserver:unknown:9::Error fetching METAR for KJFK\r\n"
if len(packets) != 1 {
t.Errorf("expected 1 packet sent, got %d", len(packets))

15
fsd/packet.go
View File

@@ -1,6 +1,7 @@
package fsd
import (
"github.com/renorris/openfsd/internal/session"
"github.com/renorris/openfsd/pkg/protocol"
)
@@ -55,7 +56,7 @@ func minFields(packetType PacketType) int {
return protocol.MinFields(packetType)
}
type handlerFunc func(client *Client, packet []byte)
type handlerFunc func(client *session.Session, packet []byte)
func getSourceCallsign(packet []byte, packetType PacketType) []byte {
return protocol.SourceCallsign(packet, packetType)
@@ -66,26 +67,26 @@ func verifySourceCallsign(packet []byte, packetType PacketType, callsign string)
}
// verifyPacket runs a set of sanity checks against a packet sent by a client and returns the detected packet type
func verifyPacket(packet []byte, client *Client) (packetType PacketType, ok bool) {
func verifyPacket(packet []byte, client *session.Session) (packetType PacketType, ok bool) {
numFields := countFields(packet)
if len(packet) < 8 || numFields < 3 {
client.sendError(SyntaxError, "Packet too short")
client.SendError(SyntaxError, "Packet too short")
return
}
packetType = getPacketType(packet)
if packetType == PacketTypeUnknown {
client.sendError(SyntaxError, "Unknown packet type")
client.SendError(SyntaxError, "Unknown packet type")
return
}
if !verifySourceCallsign(packet, packetType, client.callsign) {
client.sendError(SourceInvalidError, "Source invalid")
if !verifySourceCallsign(packet, packetType, client.Callsign) {
client.SendError(SourceInvalidError, "Source invalid")
return
}
if numFields < minFields(packetType) {
client.sendError(SyntaxError, "Minimum field count requirement not satisfied")
client.SendError(SyntaxError, "Minimum field count requirement not satisfied")
return
}

89
fsd/postoffice.go
View File

@@ -6,22 +6,23 @@ import (
"sync"
"github.com/renorris/openfsd/internal/geo"
"github.com/renorris/openfsd/internal/session"
"github.com/tidwall/rtree"
)
type postOffice struct {
clientMap map[string]*Client // Callsign -> *Client
clientMap map[string]*session.Session // Callsign -> *session.Session
clientMapLock *sync.RWMutex
tree *rtree.RTreeG[*Client] // Geospatial rtree
tree *rtree.RTreeG[*session.Session] // Geospatial rtree
treeLock *sync.RWMutex
}
func newPostOffice() *postOffice {
return &postOffice{
clientMap: make(map[string]*Client, 128),
clientMap: make(map[string]*session.Session, 128),
clientMapLock: &sync.RWMutex{},
tree: &rtree.RTreeG[*Client]{},
tree: &rtree.RTreeG[*session.Session]{},
treeLock: &sync.RWMutex{},
}
}
@@ -29,49 +30,49 @@ func newPostOffice() *postOffice {
var ErrCallsignInUse = errors.New("callsign in use")
var ErrCallsignDoesNotExist = errors.New("callsign does not exist")
// register adds a new Client to the post office. Returns ErrCallsignInUse when the callsign is taken.
func (p *postOffice) register(client *Client) (err error) {
// register adds a new Session to the post office. Returns ErrCallsignInUse when the callsign is taken.
func (p *postOffice) register(s *session.Session) (err error) {
p.clientMapLock.Lock()
if _, exists := p.clientMap[client.callsign]; exists {
if _, exists := p.clientMap[s.Callsign]; exists {
p.clientMapLock.Unlock()
err = ErrCallsignInUse
return
}
p.clientMap[client.callsign] = client
p.clientMap[s.Callsign] = s
p.clientMapLock.Unlock()
// Insert into R-tree
clientMin, clientMax := geo.BoundingBox(client.latLon(), client.visRange.Load())
clientMin, clientMax := geo.BoundingBox(s.LatLon(), s.VisRange.Load())
p.treeLock.Lock()
p.tree.Insert(clientMin, clientMax, client)
p.tree.Insert(clientMin, clientMax, s)
p.treeLock.Unlock()
return
}
// release removes a Client from the post office.
func (p *postOffice) release(client *Client) {
clientMin, clientMax := geo.BoundingBox(client.latLon(), client.visRange.Load())
// release removes a Session from the post office.
func (p *postOffice) release(s *session.Session) {
clientMin, clientMax := geo.BoundingBox(s.LatLon(), s.VisRange.Load())
p.treeLock.Lock()
p.tree.Delete(clientMin, clientMax, client)
p.tree.Delete(clientMin, clientMax, s)
p.treeLock.Unlock()
p.clientMapLock.Lock()
delete(p.clientMap, client.callsign)
delete(p.clientMap, s.Callsign)
p.clientMapLock.Unlock()
return
}
// updatePosition updates the geospatial position of a Client.
// The referenced client's latLon and visRange are rewritten.
func (p *postOffice) updatePosition(client *Client, newCenter [2]float64, newVisRange float64) {
oldMin, oldMax := geo.BoundingBox(client.latLon(), client.visRange.Load())
// updatePosition updates the geospatial position of a Session.
// The referenced session's lat/lon and visRange are rewritten.
func (p *postOffice) updatePosition(s *session.Session, newCenter [2]float64, newVisRange float64) {
oldMin, oldMax := geo.BoundingBox(s.LatLon(), s.VisRange.Load())
newMin, newMax := geo.BoundingBox(newCenter, newVisRange)
client.setLatLon(newCenter[0], newCenter[1])
client.visRange.Store(newVisRange)
s.SetLatLon(newCenter[0], newCenter[1])
s.VisRange.Store(newVisRange)
// Avoid redundant updates
if oldMin == newMin && oldMax == newMax {
@@ -79,38 +80,38 @@ func (p *postOffice) updatePosition(client *Client, newCenter [2]float64, newVis
}
p.treeLock.Lock()
p.tree.Delete(oldMin, oldMax, client)
p.tree.Insert(newMin, newMax, client)
p.tree.Delete(oldMin, oldMax, s)
p.tree.Insert(newMin, newMax, s)
p.treeLock.Unlock()
return
}
// search calls `callback` for every other Client within geographical range of the provided Client.
// search calls `callback` for every other Session within geographical range of the provided Session.
//
// It resets Client.closestVelocityClientDistance to +Inf, then updates it to the
// It resets Session.ClosestVelocityClientDistance to +Inf, then updates it to the
// minimum geo.Distance among non-self proto-101 pilot pairs discovered during the search.
func (p *postOffice) search(client *Client, callback func(recipient *Client) bool) {
clientMin, clientMax := geo.BoundingBox(client.latLon(), client.visRange.Load())
func (p *postOffice) search(s *session.Session, callback func(recipient *session.Session) bool) {
clientMin, clientMax := geo.BoundingBox(s.LatLon(), s.VisRange.Load())
client.closestVelocityClientDistance = math.MaxFloat64
s.ClosestVelocityClientDistance = math.MaxFloat64
p.treeLock.RLock()
p.tree.Search(clientMin, clientMax, func(foundMin [2]float64, foundMax [2]float64, foundClient *Client) bool {
if foundClient == client {
p.tree.Search(clientMin, clientMax, func(foundMin [2]float64, foundMax [2]float64, found *session.Session) bool {
if found == s {
return true // Ignore self
}
if !client.isAtc && client.protoRevision == 101 && foundClient.protoRevision == 101 {
clientLatLon := client.latLon()
foundClientLatLon := foundClient.latLon()
dist := geo.Distance(clientLatLon[0], clientLatLon[1], foundClientLatLon[0], foundClientLatLon[1])
if dist < client.closestVelocityClientDistance {
client.closestVelocityClientDistance = dist
if !s.IsAtc && s.ProtoRevision == 101 && found.ProtoRevision == 101 {
clientLatLon := s.LatLon()
foundLatLon := found.LatLon()
dist := geo.Distance(clientLatLon[0], clientLatLon[1], foundLatLon[0], foundLatLon[1])
if dist < s.ClosestVelocityClientDistance {
s.ClosestVelocityClientDistance = dist
}
}
return callback(foundClient)
return callback(found)
})
p.treeLock.RUnlock()
}
@@ -120,7 +121,7 @@ func (p *postOffice) search(client *Client, callback func(recipient *Client) boo
// Returns ErrCallsignDoesNotExist if the callsign does not exist.
func (p *postOffice) send(callsign string, packet string) (err error) {
p.clientMapLock.RLock()
client, exists := p.clientMap[callsign]
s, exists := p.clientMap[callsign]
p.clientMapLock.RUnlock()
if !exists {
@@ -128,15 +129,15 @@ func (p *postOffice) send(callsign string, packet string) (err error) {
return
}
return client.send(packet)
return s.Send(packet)
}
// find finds a Client with a given callsign.
// find finds a Session with a given callsign.
//
// Returns ErrCallsignDoesNotExist if the callsign does not exist.
func (p *postOffice) find(callsign string) (client *Client, err error) {
func (p *postOffice) find(callsign string) (s *session.Session, err error) {
p.clientMapLock.RLock()
client, exists := p.clientMap[callsign]
s, exists := p.clientMap[callsign]
p.clientMapLock.RUnlock()
if !exists {
@@ -147,10 +148,10 @@ func (p *postOffice) find(callsign string) (client *Client, err error) {
}
// all calls `callback` for every single client registered to the post office.
func (p *postOffice) all(client *Client, callback func(recipient *Client) bool) {
func (p *postOffice) all(s *session.Session, callback func(recipient *session.Session) bool) {
p.clientMapLock.RLock()
for _, recipient := range p.clientMap {
if recipient == client {
if recipient == s {
continue
}
if !callback(recipient) {

114
fsd/postoffice_test.go
View File

@@ -11,13 +11,14 @@ import (
"testing"
"github.com/renorris/openfsd/internal/geo"
"github.com/renorris/openfsd/internal/session"
)
func newTestClient(callsign string, lat, lon, visRange float64) *Client {
c := &Client{loginData: loginData{callsign: callsign}}
c.setLatLon(lat, lon)
c.visRange.Store(visRange)
return c
func newTestClient(callsign string, lat, lon, visRange float64) *session.Session {
s := session.New(context.Background(), nil, nil, session.LoginData{Callsign: callsign})
s.SetLatLon(lat, lon)
s.VisRange.Store(visRange)
return s
}
// TestRegister tests the registration of clients with unique and duplicate callsigns.
@@ -53,8 +54,8 @@ func TestRelease(t *testing.T) {
t.Fatal(err)
}
var found []*Client
p.search(client2, func(recipient *Client) bool {
var found []*session.Session
p.search(client2, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
@@ -68,7 +69,7 @@ func TestRelease(t *testing.T) {
}
found = nil
p.search(client2, func(recipient *Client) bool {
p.search(client2, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
@@ -89,8 +90,8 @@ func TestUpdatePosition(t *testing.T) {
t.Fatal(err)
}
var found []*Client
p.search(client1, func(recipient *Client) bool {
var found []*session.Session
p.search(client1, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
@@ -101,7 +102,7 @@ func TestUpdatePosition(t *testing.T) {
p.updatePosition(client2, [2]float64{100.0, 100.0}, 100000)
found = nil
p.search(client1, func(recipient *Client) bool {
p.search(client1, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
@@ -126,17 +127,17 @@ func TestUpdatePosition_NoopKeepsIndexed(t *testing.T) {
// Same center and range → identical bbox → early return (no tree rewrite).
p.updatePosition(client1, [2]float64{10, 20}, 50000)
latLon := client1.latLon()
latLon := client1.LatLon()
if latLon[0] != 10 || latLon[1] != 20 {
t.Fatalf("latLon after noop update = %v", latLon)
}
if client1.visRange.Load() != 50000 {
t.Fatalf("visRange after noop update = %v", client1.visRange.Load())
if client1.VisRange.Load() != 50000 {
t.Fatalf("visRange after noop update = %v", client1.VisRange.Load())
}
// Peer must still find client1, proving the tree entry remains valid.
var found []*Client
p.search(peer, func(recipient *Client) bool {
var found []*session.Session
p.search(peer, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
@@ -161,26 +162,26 @@ func TestSearch(t *testing.T) {
t.Fatal(err)
}
var found []*Client
p.search(client1, func(recipient *Client) bool {
var found []*session.Session
p.search(client1, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
if len(found) != 1 || found[0].callsign != "client2" {
if len(found) != 1 || found[0].Callsign != "client2" {
t.Errorf("expected to find client2, got %v", found)
}
found = nil
p.search(client2, func(recipient *Client) bool {
p.search(client2, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
if len(found) != 1 || found[0].callsign != "client1" {
if len(found) != 1 || found[0].Callsign != "client1" {
t.Errorf("expected to find client1, got %v", found)
}
found = nil
p.search(client3, func(recipient *Client) bool {
p.search(client3, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
@@ -194,13 +195,13 @@ func TestSearch(t *testing.T) {
}
found = nil
p.search(client1, func(recipient *Client) bool {
p.search(client1, func(recipient *session.Session) bool {
found = append(found, recipient)
return true
})
foundCallsigns := make([]string, len(found))
for i, c := range found {
foundCallsigns[i] = c.callsign
foundCallsigns[i] = c.Callsign
}
sort.Strings(foundCallsigns)
expected := []string{"client2", "client4"}
@@ -217,33 +218,33 @@ func TestSearch(t *testing.T) {
}
// TestSearch_ClosestVelocityDistance ensures proto-101 pilot pairs update
// closestVelocityClientDistance via geo.Distance.
// ClosestVelocityClientDistance via geo.Distance.
func TestSearch_ClosestVelocityDistance(t *testing.T) {
p := newPostOffice()
client1 := newTestClient("v1", 0, 0, 500000)
client1.protoRevision = 101
client1.isAtc = false
client1.ProtoRevision = 101
client1.IsAtc = false
if err := p.register(client1); err != nil {
t.Fatal(err)
}
client2 := newTestClient("v2", 0.1, 0, 500000)
client2.protoRevision = 101
client2.isAtc = false
client2.ProtoRevision = 101
client2.IsAtc = false
if err := p.register(client2); err != nil {
t.Fatal(err)
}
// Non-101 neighbor should not affect closest velocity distance.
client3 := newTestClient("old", 0.05, 0, 500000)
client3.protoRevision = 100
client3.ProtoRevision = 100
if err := p.register(client3); err != nil {
t.Fatal(err)
}
p.search(client1, func(recipient *Client) bool { return true })
p.search(client1, func(recipient *session.Session) bool { return true })
want := geo.Distance(0, 0, 0.1, 0)
if !approxEqual(client1.closestVelocityClientDistance, want) {
t.Fatalf("closestVelocityClientDistance = %v, want %v", client1.closestVelocityClientDistance, want)
if !approxEqual(client1.ClosestVelocityClientDistance, want) {
t.Fatalf("ClosestVelocityClientDistance = %v, want %v", client1.ClosestVelocityClientDistance, want)
}
}
@@ -272,15 +273,15 @@ func TestAll(t *testing.T) {
self := newTestClient("self", 0, 0, 1000)
a := newTestClient("a", 0, 0, 1000)
b := newTestClient("b", 0, 0, 1000)
for _, c := range []*Client{self, a, b} {
for _, c := range []*session.Session{self, a, b} {
if err := p.register(c); err != nil {
t.Fatal(err)
}
}
var seen []string
p.all(self, func(recipient *Client) bool {
seen = append(seen, recipient.callsign)
p.all(self, func(recipient *session.Session) bool {
seen = append(seen, recipient.Callsign)
return true
})
sort.Strings(seen)
@@ -290,7 +291,7 @@ func TestAll(t *testing.T) {
// Early stop: callback returns false.
count := 0
p.all(self, func(recipient *Client) bool {
p.all(self, func(recipient *session.Session) bool {
count++
return false
})
@@ -305,9 +306,9 @@ func TestSend(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
client := newTestClient("RECV", 0, 0, 1000)
client.ctx = ctx
client.sendChan = make(chan string, 1)
client := session.New(ctx, nil, nil, session.LoginData{Callsign: "RECV"})
client.SetLatLon(0, 0)
client.VisRange.Store(1000)
if err := p.register(client); err != nil {
t.Fatal(err)
}
@@ -315,14 +316,13 @@ func TestSend(t *testing.T) {
if err := p.send("RECV", "hello\r\n"); err != nil {
t.Fatalf("send existing: %v", err)
}
select {
case pkt := <-client.sendChan:
if pkt != "hello\r\n" {
t.Fatalf("packet = %q, want hello\\r\\n", pkt)
}
default:
pkt, ok := client.DequeueOutbound()
if !ok {
t.Fatal("expected packet on sendChan")
}
if pkt != "hello\r\n" {
t.Fatalf("packet = %q, want hello\\r\\n", pkt)
}
if err := p.send("NOPE", "x"); err != ErrCallsignDoesNotExist {
t.Fatalf("send missing: err = %v, want ErrCallsignDoesNotExist", err)
@@ -339,7 +339,7 @@ func approxEqual(a, b float64) bool {
func TestPostOfficeConcurrent(t *testing.T) {
p := newPostOffice()
const n = 64
clients := make([]*Client, n)
clients := make([]*session.Session, n)
for i := 0; i < n; i++ {
clients[i] = newTestClient(fmt.Sprintf("C%d", i), float64(i%10), float64(i%20), 200000)
}
@@ -348,10 +348,10 @@ func TestPostOfficeConcurrent(t *testing.T) {
// Register all.
for i := 0; i < n; i++ {
wg.Add(1)
go func(c *Client) {
go func(c *session.Session) {
defer wg.Done()
if err := p.register(c); err != nil {
t.Errorf("register %s: %v", c.callsign, err)
t.Errorf("register %s: %v", c.Callsign, err)
}
}(clients[i])
}
@@ -360,12 +360,12 @@ func TestPostOfficeConcurrent(t *testing.T) {
// Concurrent search + updatePosition.
for i := 0; i < n; i++ {
wg.Add(1)
go func(c *Client) {
go func(c *session.Session) {
defer wg.Done()
p.search(c, func(recipient *Client) bool { return true })
ll := c.latLon()
p.search(c, func(recipient *session.Session) bool { return true })
ll := c.LatLon()
p.updatePosition(c, [2]float64{ll[0] + 0.01, ll[1] - 0.01}, 150000)
p.search(c, func(recipient *Client) bool { return true })
p.search(c, func(recipient *session.Session) bool { return true })
}(clients[i])
}
wg.Wait()
@@ -373,7 +373,7 @@ func TestPostOfficeConcurrent(t *testing.T) {
// Concurrent release.
for i := 0; i < n; i++ {
wg.Add(1)
go func(c *Client) {
go func(c *session.Session) {
defer wg.Done()
p.release(c)
}(clients[i])
@@ -388,7 +388,7 @@ func TestPostOfficeConcurrent(t *testing.T) {
// BenchmarkRegister measures client registration cost.
func BenchmarkRegister(b *testing.B) {
r := rand.New(rand.NewSource(42))
clients := make([]*Client, b.N)
clients := make([]*session.Session, b.N)
for i := 0; i < b.N; i++ {
clients[i] = newTestClient(
fmt.Sprintf("C%d", i),
@@ -413,7 +413,7 @@ func benchmarkSearchWithN(b *testing.B, n int) {
p := newPostOffice()
r := rand.New(rand.NewSource(42))
clients := make([]*Client, n)
clients := make([]*session.Session, n)
for i := 0; i < n; i++ {
clients[i] = newTestClient(
fmt.Sprintf("Client%d", i),
@@ -426,7 +426,7 @@ func benchmarkSearchWithN(b *testing.B, n int) {
}
}
callback := func(recipient *Client) bool {
callback := func(recipient *session.Session) bool {
return true
}

63
fsd/util.go
View File

@@ -8,6 +8,7 @@ import (
"strconv"
"strings"
"github.com/renorris/openfsd/internal/session"
"github.com/renorris/openfsd/pkg/protocol"
)
@@ -194,11 +195,11 @@ func parseVisRange(packet []byte, index int) (visRange float64, ok bool) {
}
// forwardClientQuery freely routes a client query packet depending on the recipient.
func forwardClientQuery(po *postOffice, client *Client, packet []byte) {
func forwardClientQuery(po *postOffice, client *session.Session, packet []byte) {
recipient := getField(packet, 1)
if len(recipient) < 2 {
client.sendError(NoSuchCallsignError, "Invalid recipient")
client.SendError(NoSuchCallsignError, "Invalid recipient")
return
}
@@ -217,68 +218,68 @@ func forwardClientQuery(po *postOffice, client *Client, packet []byte) {
}
// broadcastRanged broadcasts a packet to all clients in range
func broadcastRanged(po *postOffice, client *Client, packet []byte) {
func broadcastRanged(po *postOffice, client *session.Session, packet []byte) {
packetStr := string(packet)
po.search(client, func(recipient *Client) bool {
recipient.send(packetStr)
po.search(client, func(recipient *session.Session) bool {
recipient.Send(packetStr)
return true
})
}
// broadcastRangedVelocity broadcasts a packet to all clients in range
// supporting the Vatsim2022 (101) protocol revision.
func broadcastRangedVelocity(po *postOffice, client *Client, packet []byte) {
func broadcastRangedVelocity(po *postOffice, client *session.Session, packet []byte) {
packetStr := string(packet)
po.search(client, func(recipient *Client) bool {
if recipient.protoRevision != 101 {
po.search(client, func(recipient *session.Session) bool {
if recipient.ProtoRevision != 101 {
return true
}
recipient.send(packetStr)
recipient.Send(packetStr)
return true
})
}
// broadcastRangedAtcOnly broadcasts a packet to all ATC clients in range
func broadcastRangedAtcOnly(po *postOffice, client *Client, packet []byte) {
func broadcastRangedAtcOnly(po *postOffice, client *session.Session, packet []byte) {
packetStr := string(packet)
po.search(client, func(recipient *Client) bool {
if !recipient.isAtc {
po.search(client, func(recipient *session.Session) bool {
if !recipient.IsAtc {
return true
}
recipient.send(packetStr)
recipient.Send(packetStr)
return true
})
}
// broadcastAll broadcasts a packet to the entire server
func broadcastAll(po *postOffice, client *Client, packet []byte) {
func broadcastAll(po *postOffice, client *session.Session, packet []byte) {
packetStr := string(packet)
po.all(client, func(recipient *Client) bool {
recipient.send(packetStr)
po.all(client, func(recipient *session.Session) bool {
recipient.Send(packetStr)
return true
})
}
// broadcastAllATC broadcasts a packet to all ATC on entire server
func broadcastAllATC(po *postOffice, client *Client, packet []byte) {
func broadcastAllATC(po *postOffice, client *session.Session, packet []byte) {
packetStr := string(packet)
po.all(client, func(recipient *Client) bool {
if !recipient.isAtc {
po.all(client, func(recipient *session.Session) bool {
if !recipient.IsAtc {
return true
}
recipient.send(packetStr)
recipient.Send(packetStr)
return true
})
}
// broadcastAll broadcasts a packet to all supervisors on the server
func broadcastAllSupervisors(po *postOffice, client *Client, packet []byte) {
func broadcastAllSupervisors(po *postOffice, client *session.Session, packet []byte) {
packetStr := string(packet)
po.all(client, func(recipient *Client) bool {
if recipient.networkRating < NetworkRatingSupervisor {
po.all(client, func(recipient *session.Session) bool {
if recipient.NetworkRating < NetworkRatingSupervisor {
return true
}
recipient.send(packetStr)
recipient.Send(packetStr)
return true
})
}
@@ -286,9 +287,9 @@ func broadcastAllSupervisors(po *postOffice, client *Client, packet []byte) {
// sendDirectOrErr attempts to send a packet directly to a recipient.
// If the post office responds with an ErrCallsignDoesNotExist, the client
// is notified with a NoSuchCallsignError.
func sendDirectOrErr(po *postOffice, client *Client, recipient []byte, packet []byte) {
func sendDirectOrErr(po *postOffice, client *session.Session, recipient []byte, packet []byte) {
if err := po.send(string(recipient), string(packet)); err != nil {
client.sendError(NoSuchCallsignError, "No such callsign")
client.SendError(NoSuchCallsignError, "No such callsign")
return
}
}
@@ -380,21 +381,21 @@ func strPtr(str string) *string {
}
// sendEnableSendFastPacket sends an 'enable' $SF Send Fast packet to the client
func sendEnableSendFastPacket(client *Client) {
func sendEnableSendFastPacket(client *session.Session) {
sendSendFastPacket(client, true)
}
// sendDisableSendFastPacket sends a 'disable' $SF Send Fast packet to the client
func sendDisableSendFastPacket(client *Client) {
func sendDisableSendFastPacket(client *session.Session) {
sendSendFastPacket(client, false)
}
// sendSendFastPacket sends a $SF Send Fast packet to the client
func sendSendFastPacket(client *Client, enabled bool) {
func sendSendFastPacket(client *session.Session, enabled bool) {
builder := strings.Builder{}
builder.Grow(32)
builder.WriteString("$SFSERVER:")
builder.WriteString(client.callsign)
builder.WriteString(client.Callsign)
builder.WriteByte(':')
if enabled {
builder.WriteByte('1')
@@ -403,5 +404,5 @@ func sendSendFastPacket(client *Client, enabled bool) {
}
builder.WriteString("\r\n")
client.send(builder.String())
client.Send(builder.String())
}

188
internal/session/session.go Normal file
View File

@@ -0,0 +1,188 @@
// Package session owns a connected FSD participant after login.
//
// This package must not import postoffice, the fsd server package, or any
// higher-level orchestration — fsd depends on session, not the reverse.
package session
import (
"bufio"
"context"
"net"
"time"
"github.com/renorris/openfsd/pkg/protocol"
"go.uber.org/atomic"
)
// Sender is anything that can enqueue outbound FSD text.
type Sender interface {
Send(packet string) error
}
// Auth is the optional VATSIM client-auth challenge state for a session.
// Implemented by fsd's vatsimAuthState (auth remains in fsd until PR4).
type Auth interface {
Initialize(clientID uint16, initialChallenge []byte) error
IsInitialized() bool
GetResponseForChallenge(challenge []byte) [32]byte
UpdateState(d *[32]byte)
}
// LoginData holds the data extracted from the client's login packets.
type LoginData struct {
ClientChallenge string // Optional client challenge for authentication
Callsign string // Callsign of the client
CID int // Cert ID
RealName string // Real name
NetworkRating protocol.NetworkRating // Network rating of the client
MaxNetworkRating protocol.NetworkRating // Maximum allowed network rating (from DB/JWT)
ProtoRevision int // Protocol revision
LoginTime time.Time // Time of login
ClientID uint16 // Client ID (from ident packet)
IsAtc bool // True if the client is ATC, false if a pilot
}
// LatLon is a geographic coordinate pair stored in Session.Coords.
type LatLon struct {
Lat, Lon float64
}
// Session is a connected FSD participant after successful login.
//
// # Field ownership
//
// Read-loop only (eventLoop / packet handlers on this connection's goroutine;
// do not read or write from other goroutines without additional sync):
// - FacilityType, SendFastEnabled, ClosestVelocityClientDistance
// - Auth (Initialize / challenge handling)
// - Scanner (owned exclusively by the read loop)
// - Conn for RemoteAddr-style metadata reads from the read loop
//
// Atomic / concurrent-safe (may be read by postoffice, HTTP service, or other
// session goroutines; writers are typically the owning read loop or postoffice
// position updates):
// - Coords (via LatLon/SetLatLon), VisRange
// - FlightPlan, AssignedBeaconCode
// - Frequency, Altitude, Groundspeed, Transponder, Heading, LastUpdated
//
// Immutable after login (set during login; safe to read concurrently afterward):
// - LoginData fields (Callsign, CID, RealName, NetworkRating, ProtoRevision, …)
// - MaxNetworkRating is fixed once authentication completes
//
// Context / outbound path:
// - Ctx, Cancel — lifecycle; Cancel is safe from any goroutine
// - sendChan — private; producers must call Send; only SenderWorker writes to Conn
//
// # Outbound I/O rule
//
// After login, all packet writes to the client MUST go through Send → sendChan →
// SenderWorker. Direct Conn.Write outside SenderWorker is forbidden post-login
// (login-phase errors may still use protocol.WriteError on the raw connection
// before SenderWorker is started).
type Session struct {
Conn net.Conn
Scanner *bufio.Scanner
Ctx context.Context
Cancel context.CancelFunc
sendChan chan string
// Coords stores LatLon; use LatLon/SetLatLon.
Coords atomic.Value
VisRange atomic.Float64
ClosestVelocityClientDistance float64 // Closest Velocity-compatible client distance in meters
FlightPlan atomic.String
AssignedBeaconCode atomic.String
Frequency atomic.String // ATC frequency
Altitude atomic.Int32 // Pilot altitude
Groundspeed atomic.Int32 // Pilot ground speed
Transponder atomic.String // Active pilot transponder
Heading atomic.Int32 // Pilot heading
LastUpdated atomic.Time // Last position/state update time
FacilityType int // ATC facility type (ATC only)
LoginData
Auth Auth // Optional; set by fsd when client auth is used
SendFastEnabled bool
}
// New constructs a Session with a cancellable child context and outbound buffer.
// conn may be nil in unit tests that only exercise Send/state.
func New(ctx context.Context, conn net.Conn, scanner *bufio.Scanner, data LoginData) *Session {
sessionCtx, cancel := context.WithCancel(ctx)
s := &Session{
Conn: conn,
Scanner: scanner,
Ctx: sessionCtx,
Cancel: cancel,
sendChan: make(chan string, 32),
LoginData: data,
}
s.SetLatLon(0, 0)
return s
}
// SenderWorker drains sendChan and writes packets to Conn until the context ends
// or a write fails. It is the only post-login code path allowed to Conn.Write.
// On exit it closes Conn and cancels the session context.
func (s *Session) SenderWorker() {
if s.Conn != nil {
defer s.Conn.Close()
}
defer s.Cancel()
for {
select {
case packet := <-s.sendChan:
if s.Conn == nil {
continue
}
if _, err := s.Conn.Write([]byte(packet)); err != nil {
return
}
case <-s.Ctx.Done():
return
}
}
}
// SendError enqueues an FSD $ER packet via the outbound send channel.
// Thread-safe; must only be used after SenderWorker is running (post-login).
func (s *Session) SendError(code int, message string) error {
return s.Send(protocol.FormatError(protocol.ErrorCode(code), message))
}
// Send queues a packet on the session's outbound channel.
// Blocks until the packet is queued or the session context is done.
func (s *Session) Send(packet string) error {
select {
case s.sendChan <- packet:
return nil
case <-s.Ctx.Done():
return s.Ctx.Err()
}
}
// LatLon returns the current [lat, lon] coordinates.
func (s *Session) LatLon() [2]float64 {
ll := s.Coords.Load().(LatLon)
return [2]float64{ll.Lat, ll.Lon}
}
// SetLatLon stores the current coordinates atomically.
func (s *Session) SetLatLon(lat, lon float64) {
s.Coords.Store(LatLon{Lat: lat, Lon: lon})
}
// DequeueOutbound non-blockingly takes one queued outbound packet.
// Intended for unit tests that assert on enqueued wire text without a Conn.
func (s *Session) DequeueOutbound() (packet string, ok bool) {
select {
case packet = <-s.sendChan:
ok = true
default:
}
return
}