gateway

package
v0.1.20 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 1, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Overview

Package gateway is the message pipeline: it turns bytes from a peer into changes on a record, and changes on a record into bytes for a peer.

The pipeline has one non-negotiable rule. Raw bytes are made durable before anything interprets them, and every later stage is a pure function of those bytes plus configuration. That is what makes replay possible after a parser fix, and it is why a decode failure produces a message sitting in the dead letter queue rather than a booking that never happened.

Index

Constants

This section is empty.

Variables

View Source 
var ErrNotAssociable = errors.New("gateway: cannot associate a value coupon to a segment with no flight coupon")

ErrNotAssociable is returned when an EMD-A names a segment that has no flight coupon to associate to.

View Source 
var ErrNotDivided = errors.New("gateway: record is not part of a division")

ErrNotDivided is returned when a locator names no divided record.

View Source 
var ErrNothingToCancel = errors.New("gateway: record has no live segment to cancel")

ErrNothingToCancel is returned when a record has no live segment.

View Source 
var ErrNothingToTicket = errors.New("gateway: record has no live air segment to ticket")

ErrNothingToTicket is returned when a record has no segment a ticket could cover.

Functions

func NewMessageID 

func NewMessageID() string

NewMessageID mints an identifier for a message that will enter the pipeline later, so a spooled message keeps one identity from the moment it lands on disk through to the store.

func TicketSummary 

func TicketSummary(rec *pnr.PNR) string

TicketSummary renders a record's ticketing state for display.

func TrimForLog 

func TrimForLog(raw []byte, n int) string

TrimForLog shortens a raw message for a log line.

Types

type BookResult 

type BookResult struct {
	PNR      *pnr.PNR `json:"pnr"`
	Sent     []string `json:"sent"`
	Carriers []string `json:"carriers"`
}

BookResult reports what a booking produced.

type BookingPassenger 

type BookingPassenger struct {
	Surname string `json:"surname"`
	Given   string `json:"given"`
	Title   string `json:"title,omitempty"`
	Infant  bool   `json:"infant,omitempty"`
}

BookingPassenger is one traveller on a booking request.

type BookingRequest 

type BookingRequest struct {
	Passengers   []BookingPassenger `json:"passengers"`
	Segments     []BookingSegment   `json:"segments"`
	SSRs         []BookingSSR       `json:"ssrs,omitempty"`
	Contact      string             `json:"contact,omitempty"`
	ReceivedFrom string             `json:"received_from,omitempty"`
	Agent        string             `json:"agent,omitempty"`
	// Channel says where the booking came from: the console API, an NDC order,
	// a partner message. It is carried onto the record and onto the span,
	// because "how much comes through NDC" is a question somebody asks.
	Channel string `json:"channel,omitempty"`
}

BookingRequest is what the distribution side receives from an agent.

func (*BookingRequest) Validate 

func (r *BookingRequest) Validate() error

Validate checks a request is coherent before anything is written.

type BookingSSR 

type BookingSSR struct {
	Code    string `json:"code"`
	Carrier string `json:"carrier,omitempty"`
	Text    string `json:"text,omitempty"`
}

BookingSSR is a special service request on a booking.

type BookingSegment 

type BookingSegment struct {
	Carrier   string `json:"carrier"`
	FlightNum string `json:"flight_num"`
	Class     string `json:"class"`
	// Date is DDMMM as an agent would enter it, e.g. "15JUN".
	Date  string `json:"date"`
	Board string `json:"board"`
	Off   string `json:"off"`
	Seats int    `json:"seats"`
	// DepartTime and ArriveTime are HHMM local to their station. Optional: a
	// sell is valid without them, and the carrier's own schedule is
	// authoritative either way.
	DepartTime string `json:"depart_time,omitempty"`
	ArriveTime string `json:"arrive_time,omitempty"`
}

BookingSegment is one requested flight.

type Bus 

type Bus struct {
	// Now, when set, stamps events instead of the wall clock, so a driven
	// simulation's bus timeline matches its stores. Set it before Publish is
	// first called; it is read without a lock.
	Now func() time.Time
	// contains filtered or unexported fields
}

Bus fans events out to live observers and keeps a short backlog so a console opened mid-flight can render recent history instead of an empty screen.

Subscribers get a buffered channel and are dropped from a send if they fall behind. A slow browser tab must never be able to stall message processing.

func NewBus 

func NewBus(n int) *Bus

NewBus returns a bus retaining the last n events.

func (*Bus) History 

func (b *Bus) History() []Event

History returns the retained backlog.

func (*Bus) Publish 

func (b *Bus) Publish(t EventType, data any)

Publish records an event and delivers it to current subscribers.

func (*Bus) Subscribe 

func (b *Bus) Subscribe() (<-chan Event, func())

Subscribe returns a channel of events and a function to release it.

type CancelOptions 

type CancelOptions struct {
	// Segments names the segments to cancel by record position. Nil cancels
	// every live segment, which is what cancelling a booking means.
	Segments []int
	By       string
	Reason   string
}

CancelOptions controls a cancellation.

type CancelResult 

type CancelResult struct {
	PNR         *pnr.PNR
	Notified    []string
	Unreachable []string
}

CancelResult reports what a cancellation actually achieved.

Notified and Unreachable are separate because they are different facts and the difference matters: a carrier we could not tell is still holding seats for a booking this node now shows as cancelled.

type Delivery 

type Delivery struct {
	Address string `json:"address"`
	// Peer is the link the address resolved to, empty when it did not resolve.
	Peer string `json:"peer,omitempty"`
	// MessageID is the outbound log entry, empty when nothing was sent.
	MessageID string `json:"message_id,omitempty"`
	// Self reports that the address is this node, so the message terminates
	// here rather than being forwarded.
	Self bool `json:"self,omitempty"`
	// Err is why an addressee was not reached.
	Err string `json:"error,omitempty"`
}

Delivery is what happened to one addressee of a fanned-out message.

type EMDCoupon 

type EMDCoupon struct {
	// RFISC is what specifically was bought. Mandatory: a coupon with no
	// sub-code says a fee was charged without saying what for.
	RFISC string
	// SegmentRef is the segment whose flight coupon this value coupon is
	// associated to. Required for an EMD-A, meaningless on an EMD-S.
	SegmentRef int
	Amount     string
	Currency   string
	// ConsumedAtIssuance puts the coupon straight to a final status, for a
	// service delivered at the counter rather than in the air.
	ConsumedAtIssuance bool
}

EMDCoupon is one value coupon to issue.

type EMDRequest 

type EMDRequest struct {
	Locator string
	PaxRef  int
	Type    pnr.DocumentType
	RFIC    pnr.RFIC
	// AirlineCode is the three-digit numeric stock code, as for a ticket.
	AirlineCode string
	IssuedBy    string
	Coupons     []EMDCoupon
}

EMDRequest asks for a miscellaneous document.

type Event 

type Event struct {
	Seq  int64     `json:"seq"`
	Type EventType `json:"type"`
	At   time.Time `json:"at"`
	Data any       `json:"data"`
}

Event is one thing worth telling an observer about.

type EventType 

type EventType string

EventType names a bus event. 

const (
	// EvMessage carries a message that entered or left the gateway.
	EvMessage EventType = "message"
	// EvPNR carries a record whose state changed.
	EvPNR EventType = "pnr"
	// EvLink reports a peer connecting or disconnecting.
	EvLink EventType = "link"
	// EvAvail reports that availability changed.
	EvAvail EventType = "avail"
	// EvQueue reports a record placed on a work queue.
	EvQueue EventType = "queue"
	// EvMovement reports an aircraft movement message: a departure, an
	// arrival, a delay, a diversion. It is what an operations display draws
	// planes from.
	EvMovement EventType = "movement"
	// EvTrace narrates a pipeline step, which is what makes the console show
	// the path a booking took rather than only its result.
	EvTrace EventType = "trace"
)

type Gateway 

type Gateway struct {
	Identity Identity
	Store    store.Store
	Bus      *Bus
	Log      *slog.Logger
	Sender   Sender

	// Responder decides how to answer an inbound request. Nil means this node
	// does not answer requests, which is the right behaviour for a distribution
	// system that only originates them.
	Responder Responder

	// Now, when set, replaces the wall clock. Every timestamp this gateway
	// stamps -- records, events, messages, references -- comes through it, so
	// a simulation can drive time and a replay can pin it. Nil reads the wall.
	Now func() time.Time

	// Avail is what this node believes is sellable. Nil disables free sale,
	// and every segment is then requested -- correct, just slower.
	Avail *avail.Cache

	// Queues turns partner answers into work. Nil means this node keeps no
	// queues, which is right for a simulated carrier and wrong for a GDS.
	Queues *queue.Manager

	// ScheduleScanLimit caps how many affected bookings one schedule message
	// may queue. It is a blast-radius limit, not a search boundary: the
	// lookup behind it searches every record regardless. Zero uses
	// defaultScheduleScanLimit.
	ScheduleScanLimit int

	// Relay makes the gateway forward messages addressed to peers other than
	// itself. It is off by default and should stay off unless the deployment is
	// meant to be a switch: a node that relays on behalf of anyone who can
	// reach it is an open relay, spending someone else's link budget under our
	// own originator address.
	Relay bool
	// contains filtered or unexported fields
}

Gateway is a message-processing node. One instance is a GDS; another, configured with a carrier identity and a Responder, is an airline system.

func New 

func New(id Identity, st store.Store, bus *Bus, log *slog.Logger, locatorSecret []byte) *Gateway

New builds a gateway.

func (*Gateway) AddPeer 

func (g *Gateway) AddPeer(p *Peer)

AddPeer registers a link.

func (*Gateway) AssociateEMD 

func (g *Gateway) AssociateEMD(ctx context.Context, locator string, number pnr.TicketNumber,
	coupon int, segmentRef int, by string) (*pnr.PNR, error)

AssociateEMD links a value coupon to a flight coupon, or breaks the link.

Association is per coupon, not per document, because the reasons for it come and go per coupon: a passenger checks in without the excess baggage they paid for, and that one coupon needs unstapling while the rest of the document stands.

func (*Gateway) Book 

func (g *Gateway) Book(ctx context.Context, req *BookingRequest) (*BookResult, error)

Book creates a record and requests its segments from the carriers that own them.

The record is written before any message goes out. If a link is down, the booking exists at HN and the request can be retried; the alternative, sending first and storing after, produces seats sold against a record that does not exist.

func (*Gateway) Cancel 

func (g *Gateway) Cancel(ctx context.Context, locator string, opts CancelOptions) (*CancelResult, error)

Cancel withdraws segments and tells the carriers holding them.

The order is deliberate and matches the reply path: the decision is recorded before it is sent, so this node's state and the message it sends can never disagree. The failure that leaves behind is visible rather than silent -- a carrier that could not be told lands on the divergence queue, because the booking now looks settled on this side and is not.

This is the operation whose absence blocked three separate things: NDC order cancellation, auto-cancel on a ticketing time limit, and cancelling from the console. None of them could be built while there was no way to say "off" to a carrier.

func (*Gateway) CancelExpired 

func (g *Gateway) CancelExpired(ctx context.Context, locator, reason string) ([]string, error)

CancelExpired cancels a booking whose ticketing time limit has passed.

It satisfies queue.Canceller. The unreachable carriers are returned rather than swallowed: a sweeper that reported success while a carrier still held the seats would be quietly manufacturing the divergence this whole path exists to avoid.

func (*Gateway) Fanout 

func (g *Gateway) Fanout(ctx context.Context, tb *typeb.Message, raw []byte,
	kind, pnrID, correlationID string) []Delivery

Fanout delivers one message to every address on its priority line.

A Type B message may carry several addressees and the network is expected to deliver a copy to each. Routing on the configured peer name alone cannot do that: it can only ever reach the one link a message was handed to.

The bytes are sent unchanged. Rewriting the address line per recipient would make each copy a different message from the one in the log, and the address line is part of what a partner may check.

func (*Gateway) Ingest 

func (g *Gateway) Ingest(ctx context.Context, peerName string, raw []byte) (*Result, error)

Ingest is the inbound pipeline.

The ordering here is the contract: capture, then classify, then decode, then deduplicate, then apply, then respond. Capture happens first and unconditionally so that nothing after it can lose the message.

func (*Gateway) IngestWith 

func (g *Gateway) IngestWith(ctx context.Context, peerName string, raw []byte, opts IngestOptions) (*Result, error)

IngestWith is Ingest with per-message options.

func (*Gateway) IsSelf 

func (g *Gateway) IsSelf(addr string) bool

IsSelf reports whether an address belongs to this node.

func (*Gateway) IssueEMD 

func (g *Gateway) IssueEMD(ctx context.Context, req EMDRequest) (*pnr.PNR, pnr.Ticket, error)

IssueEMD issues an electronic miscellaneous document.

An EMD-A is stapled to flight coupons and lifted with them, so every value coupon has to name a segment that is actually ticketed. An EMD-S is standalone and names none. Refusing the mismatch here is what stops a document existing that says two contradictory things about itself.

func (*Gateway) IssueTickets 

func (g *Gateway) IssueTickets(ctx context.Context, locator string, opts IssueOptions) (*pnr.PNR, error)

IssueTickets issues documents against a record, one set per passenger.

Issuing is what a ticketing time limit is waiting for, so it is also what satisfies it: the deadline is cleared and any ticketing task on the queue is worked. Leaving the limit standing after issuance would have the sweeper raise a record that has already been dealt with, every pass, forever.

func (*Gateway) Peer 

func (g *Gateway) Peer(name string) *Peer

Peer returns a configured peer by link name.

func (*Gateway) PeerByAddress 

func (g *Gateway) PeerByAddress(addr string) *Peer

PeerByAddress resolves a Type B address to the link that serves it.

func (*Gateway) PeerForCarrier 

func (g *Gateway) PeerForCarrier(carrier string) *Peer

PeerForCarrier returns the link serving a carrier's segments.

func (*Gateway) Peers 

func (g *Gateway) Peers() []*Peer

Peers returns every configured peer.

func (*Gateway) RequestFromCarrier 

func (g *Gateway) RequestFromCarrier(ctx context.Context, rec *pnr.PNR, carrier string) (string, error)

RequestFromCarrier sends a sell request for the record's segments operated by carrier, in whatever format that carrier's link speaks.

func (*Gateway) Send 

func (g *Gateway) Send(ctx context.Context, peer *Peer, raw []byte, kind, pnrID, correlationID string) (string, error)

Send records and transmits an outbound message.

Capture precedes transmission for the same reason it does inbound: a message that went out must be in the log even if recording what happened to it fails.

func (*Gateway) SendKeyed 

func (g *Gateway) SendKeyed(ctx context.Context, peer *Peer, raw []byte, kind, pnrID, correlationID, key string) (string, error)

SendKeyed is Send with an application-level key recorded against the message.

For EDIFACT the key is the interchange control reference, which is what a partner's CONTRL quotes back. Without it an acknowledgement has nothing to match against and can only be filed as a divergence.

func (*Gateway) Split 

func (g *Gateway) Split(ctx context.Context, req SplitRequest) (*SplitResult, error)

Split divides passengers onto a new record.

Why both records keep the same carrier locators

A carrier holds one booking. Dividing it here does not divide it there, so until the carrier splits too, both of our records refer to the same record of theirs. Giving the child no carrier reference would be tidier and false: an agent ringing the carrier about the child would find nothing.

Why the child is written first

There is no transaction across two records. Whichever write happens first, the other can fail. Writing the child first means a failure leaves the passengers on both records -- visible, wrong, and recoverable. Writing the parent first would mean a failure loses them from one record without putting them on another, and nothing recovers that.

type Identity 

type Identity struct {
	// Designator is the two-character company code, e.g. a GDS code or an
	// airline designator.
	Designator string
	// TTYAddress is the seven-character Type B address messages are sent from.
	TTYAddress string
	// Name is a human label for the console.
	Name string
}

Identity is how this node names itself to partners.

type IngestOptions 

type IngestOptions struct {
	// Transport names the ingress that accepted the message, for the audit
	// trail. Empty defaults to "link".
	Transport string
	// Remote describes where it came from: an address, a certificate subject,
	// or a file path.
	Remote string
	// FromFile marks a message read from a drop directory rather than a live
	// link, which permits leniencies that would be unsafe on relayed traffic.
	FromFile bool
	// HoldReply returns a generated reply in the Result instead of sending it
	// over the peer's egress. A partner posting over HTTP and waiting on the
	// response has no egress to receive a reply on, and attempting one would
	// queue an undeliverable message for a link that does not exist.
	HoldReply bool
}

IngestOptions parameterises a single inbound message.

type Inventory 

type Inventory struct {

	// Carrier is the designator this inventory holds seats for. Segments for
	// other carriers on a shared record are not this system's to answer.
	Carrier string

	// Capacity is the seats offered per booking class on a flight. Zero uses a
	// value derived from the flight number so that different flights behave
	// differently without configuration.
	Capacity int
	// WaitlistCapacity is how many seats may be waitlisted once sold out.
	WaitlistCapacity int
	// ClosedClasses are booking classes that are never available, so a demo can
	// reliably show an unable response.
	ClosedClasses map[string]bool
	// contains filtered or unexported fields
}

Inventory is a seat inventory for a simulated carrier.

It is a stand-in for a real availability and inventory system, which is the component this project explicitly does not try to be. What matters here is that the interface a carrier plugs into is small and synchronous: given a record, return a status code per segment. A production deployment replaces this with a call into the carrier's own inventory and changes nothing else.

func NewInventory 

func NewInventory() *Inventory

NewInventory returns an inventory with demo-friendly defaults.

func (*Inventory) Availability 

func (inv *Inventory) Availability(keys []avail.Key, asOf time.Time) []avail.Entry

Availability reports what this inventory would grant for each key, as the beliefs a carrier would broadcast.

This is the carrier's side of free sale: it is publishing what it is willing to have sold without being asked, which is a commitment, not a description.

func (*Inventory) Decide 

func (inv *Inventory) Decide(ctx context.Context, p *pnr.PNR, peer *Peer) (map[string]string, error)

Decide answers a request, updating the inventory to reflect what it granted.

The decision and the seat commitment happen under one lock. Deciding first and committing later is how a simulator oversells, and it is how real systems do too.

func (*Inventory) SetCapacity added in v0.1.6 

func (inv *Inventory) SetCapacity(n int)

SetCapacity replaces the per-class seat count while the inventory may already be answering traffic, which is exactly when a harness or a demo wants to raise it. The Capacity field itself is for configuration before anything else holds a reference.

func (*Inventory) SetOverride 

func (inv *Inventory) SetOverride(key, status string)

SetOverride forces the outcome for a flight key, e.g. "BA/0175/15JUN/Y".

func (*Inventory) Snapshot 

func (inv *Inventory) Snapshot() map[string]map[string]int

Snapshot returns the current inventory state, for the console.

type IssueOptions 

type IssueOptions struct {
	// AirlineCode is the three-digit numeric code whose stock is being issued
	// against, e.g. 125 for British Airways. It is not the two-letter
	// designator, and there is no reliable mapping between the two, so it has
	// to be supplied.
	AirlineCode string
	// IssuedBy names who issued, for the audit trail.
	IssuedBy string
}

IssueOptions controls ticket issuance.

type Peer 

type Peer struct {
	// Name is the link name, used for routing and as the store's peer key.
	Name string
	// Carrier is the designator whose segments this peer is authoritative for.
	Carrier string
	// Format is the wire encoding this peer speaks.
	Format store.Format
	// SequenceWrap is the number this link's channel counter returns to after
	// its highest value. Zero means unknown, and a rollover then reads as a
	// large gap rather than being silently absorbed.
	SequenceWrap int

	// CONTRL is when to send a syntax and service report for an EDITFACT
	// interchange this peer sends: "requested" (the default) honours the
	// acknowledgement request in UNB 0031, "always", "errors" reports only
	// rejections, and "never" sends none.
	CONTRL string

	// TTYAddress is the peer's Type B address, used when Format is typeb.
	TTYAddress string
	// Addresses are further Type B addresses this link serves, beyond
	// TTYAddress. A carrier commonly has one address per department and one
	// circuit, so routing on the address needs the whole set.
	Addresses []string
	// AirimpProfile overrides the AIRIMP grammar for this link. Nil uses the
	// default profile.
	AirimpProfile *airimp.Profile
	// PadisProfile overrides the PADIS segment handlers for this link.
	PadisProfile *padis.Profile
	// SSIMProfile overrides the SSM and ASM grammar for this link.
	SSIMProfile *ssim.Profile
	// AvsProfile overrides the availability grammar and status-code meanings
	// for this link. The standard makes numeric availability bilateral, so a
	// per-link map is the normal case rather than an escape hatch.
	AvsProfile *avs.Profile
}

Peer is a configured partner link.

type Responder 

type Responder interface {
	// Decide returns a status code for each segment key on the record.
	Decide(ctx context.Context, p *pnr.PNR, peer *Peer) (map[string]string, error)
}

Responder answers an inbound request. A carrier implements this to consult its inventory; a distribution system usually leaves it nil.

type Result 

type Result struct {
	MessageID string
	PNRID     string
	Locator   string
	Status    store.Status
	Duplicate bool
	Changes   []string
	Replies   []string // ids of messages sent in response
	// Reply carries the generated response when HoldReply was set.
	Reply []byte
	Err   error
}

Result reports what processing a message did.

type Sender 

type Sender interface {
	Send(ctx context.Context, peer string, raw []byte) error
}

Sender hands an outbound message to a transport.

type SenderFunc 

type SenderFunc func(ctx context.Context, peer string, raw []byte) error

SenderFunc adapts a function to Sender.

func (SenderFunc) Send 

func (f SenderFunc) Send(ctx context.Context, peer string, raw []byte) error

type SplitRequest 

type SplitRequest struct {
	Locator string
	// Passengers names who moves, by their reference on the parent.
	Passengers []int
	By         string
	Reason     string
}

SplitRequest asks for passengers to be divided onto their own record.

type SplitResult 

type SplitResult struct {
	Parent *pnr.PNR
	Child  *pnr.PNR
	// Advised lists carriers told about the division.
	Advised []string
	// Unadvised lists carriers that still hold a single record covering both
	// halves. It is not an error; it is the state of the world until the
	// teletype divide message exists.
	Unadvised []string
}

SplitResult is what a division produced.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.