engine

package
v0.1.23 Latest Latest
Warning

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

 Go to latest Published: Nov 1, 2026 License: Apache-2.0 Imports: 87 Imported by: 0

Documentation

Overview

Package engine runs configured channels.

A channel is a source, a filter, and a set of destinations. The runtime's job is to be honest about what happened to each message: an acknowledgement should mean what the configuration says it means, a message that is filtered out should be distinguishable from one that was delivered, and a message that could not be delivered anywhere must not be reported as accepted.

Index

Constants

View Source 
const DefaultAS2MaxMessageSize = 50 << 20

DefaultAS2MaxMessageSize bounds one AS2 request when the source sets no limit.

View Source 
const DefaultHTTPMaxMessageSize = 16 << 20

DefaultHTTPMaxMessageSize bounds an inbound body.

Variables

View Source 
var DefaultDirectCertsDir string

DefaultDirectCertsDir is the server's Direct address book, read when a destination names no recipient_certs_dir. Set at startup.

View Source 
var ErrNoRouter = errors.New("this server cannot route between channels")

ErrNoRouter is returned when a channel destination is configured but nothing can resolve names.

A distinct error because it means the server was assembled wrongly rather than that the channel was configured wrongly, and the two need different people to fix them.

View Source 
var ErrNotAdmitted = errors.New("not admitted")

ErrNotAdmitted wraps the reason a received message was refused before it was processed: a tenant's rate or storage limit. The message was not read, stored or delivered, and the sender is expected to resend it - so MLLP sends no acknowledgement, HTTP answers 429, and a queue or file source leaves the message where it was.

View Source 
var ErrPGPRefused = errors.New("refused")

ErrPGPRefused marks a received message that could not be opened with PGP: the sender's problem (a wrong key, no signature, not encrypted), so an HTTP sender is told 400 and why, rather than 500.

View Source 
var PatientIndex *mpi.Store

PatientIndex is the server's patient index, set at start-up; nil when the server runs without one.

View Source 
var PriorAuths *priorauth.PriorAuthTracker

PriorAuths records what PAS destinations submitted and what the payers answered, set at start-up; nil keeps no record.

View Source 
var SoftwareVersion = "dev"

SoftwareVersion is Perfuse's version as the SFT segment of an ELR message names it; the server sets it at start-up.

View Source 
var TEFCA *tefca.Participant

TEFCA is the server's TEFCA participant, set at start-up; nil when the server does not participate.

Functions

func CompileBundleScript added in v0.1.21 

func CompileBundleScript(engine *script.Engine, name, source string) (*script.Script, error)

CompileBundleScript compiles a bundle script (config.FHIRDestination.BundleScript). The site's source is wrapped so that `bundle` is the converted Bundle, parsed, and whatever it holds when the script ends is what is sent; the wrapper sits on the source's first line, so a line number in an error is the author's.

func LoadDirectAddress added in v0.1.16 

func LoadDirectAddress(a config.DirectAddress) (cms.KeyPair, []*x509.Certificate, *x509.CertPool, error)

LoadDirectAddress reads an address's certificate, chain, key and anchors.

func RunBundleScript added in v0.1.21 

func RunBundleScript(engine *script.Engine, compiled *script.Script, b *fhir.Bundle, version fhir.Version,
	log func(level, message string)) (*fhir.Bundle, error)

RunBundleScript runs a bundle script over a converted Bundle and reads back what it left. What comes back goes through the model again (members it does not model are kept), so validation and the version written apply to the site's changes too.

Types

type AMQPSender added in v0.1.5 

type AMQPSender struct {
	// contains filtered or unexported fields
}

AMQPSender sends each message to a queue or topic.

func NewAMQPSender added in v0.1.5 

func NewAMQPSender(d config.Destination) (*AMQPSender, error)

NewAMQPSender builds the sender; it connects on the first message.

func (*AMQPSender) Close added in v0.1.5 

func (s *AMQPSender) Close() error

Close ends the connection.

func (*AMQPSender) Describe added in v0.1.5 

func (s *AMQPSender) Describe() string

Describe names the broker and address, never the credentials.

func (*AMQPSender) Send added in v0.1.5 

func (s *AMQPSender) Send(ctx context.Context, msg []byte) error

Send transfers one message and waits for the broker to accept it.

type AS2Sender added in v0.1.17 

type AS2Sender struct {
	// contains filtered or unexported fields
}

AS2Sender sends each message to a trading partner over AS2. With a synchronous receipt (the default) a message is delivered only when the partner's receipt says processed and names the digest of what was sent; anything else fails the delivery, so the engine retries it.

func NewAS2Sender added in v0.1.17 

func NewAS2Sender(d config.Destination) (*AS2Sender, error)

NewAS2Sender builds the sender, reading every key and certificate now.

func (*AS2Sender) Close added in v0.1.17 

func (s *AS2Sender) Close() error

func (*AS2Sender) Describe added in v0.1.17 

func (s *AS2Sender) Describe() string

func (*AS2Sender) Send added in v0.1.17 

func (s *AS2Sender) Send(ctx context.Context, msg []byte) error

Send posts one message and, for a synchronous receipt, checks it.

func (*AS2Sender) SendForResponse added in v0.1.17 

func (s *AS2Sender) SendForResponse(ctx context.Context, msg []byte) ([]byte, error)

SendForResponse returns the receipt's text, so a response transformer can read it.

func (*AS2Sender) SetDataType added in v0.1.17 

func (s *AS2Sender) SetDataType(t config.DataType)

SetDataType sets the payload's default content type.

type AthenaRecord added in v0.1.5 

type AthenaRecord struct {
	ReceivedAt        string `json:"received_at"`
	Channel           string `json:"channel"`
	MessageType       string `json:"message_type,omitempty"`
	TriggerEvent      string `json:"trigger_event,omitempty"`
	ControlID         string `json:"control_id,omitempty"`
	SendingApp        string `json:"sending_application,omitempty"`
	SendingFacility   string `json:"sending_facility,omitempty"`
	ReceivingFacility string `json:"receiving_facility,omitempty"`
	PatientID         string `json:"patient_id,omitempty"`
	MessageTime       string `json:"message_time,omitempty"`
	Message           string `json:"message"`
}

AthenaRecord is one archived message as Amazon Athena reads it: the fields people filter on, and the message itself.

One JSON object per S3 object, under ${channel}/dt=YYYY-MM-DD/, which is the layout `perfuse athena` writes a table for, with partition projection so no crawler or MSCK REPAIR is needed. The message is kept whole, so nothing about the archive depends on having chosen the right columns in advance.

type AzureBlobSender added in v0.1.5 

type AzureBlobSender struct {
	// contains filtered or unexported fields
}

AzureBlobSender writes each message as a blob.

func NewAzureBlobSender added in v0.1.5 

func NewAzureBlobSender(d config.Destination) (*AzureBlobSender, error)

NewAzureBlobSender builds the sender.

func (*AzureBlobSender) Close added in v0.1.5 

func (s *AzureBlobSender) Close() error

Close releases nothing.

func (*AzureBlobSender) Describe added in v0.1.5 

func (s *AzureBlobSender) Describe() string

Describe names the container, never the credential.

func (*AzureBlobSender) Send added in v0.1.5 

func (s *AzureBlobSender) Send(ctx context.Context, msg []byte) error

Send writes one blob.

func (*AzureBlobSender) SetSeal added in v0.1.17 

func (s *AzureBlobSender) SetSeal(f func([]byte) ([]byte, error))

SetSeal makes the sender encrypt each object's bytes (a pgp block).

type BrokerSender 

type BrokerSender struct {
	// contains filtered or unexported fields
}

BrokerSender publishes messages to a message broker.

Mirth's JMS Writer. The connection is opened on demand and kept, because a destination that reconnected per message would spend more time in handshakes than in sending - and a broker sees a connection per message as an attack.

func NewBrokerSender 

func NewBrokerSender(d config.Destination) (*BrokerSender, error)

NewBrokerSender builds a broker destination.

func (*BrokerSender) Close 

func (s *BrokerSender) Close() error

Close releases the connection.

func (*BrokerSender) Describe 

func (s *BrokerSender) Describe() string

Describe names the destination.

Never the login or the passcode.

func (*BrokerSender) Send 

func (s *BrokerSender) Send(ctx context.Context, msg []byte) error

Send publishes one message.

type CDASender 

type CDASender struct {
	// contains filtered or unexported fields
}

CDASender extracts the clinical document from an HL7 v2 message, converts it, and posts or writes the result.

A document arriving inside an MDM^T02 is the normal case, not an edge case, and an engine that cannot see inside that message leaves the most clinically dense payload in the feed unread.

func NewCDASender 

func NewCDASender(d config.Destination, log *slog.Logger) (*CDASender, error)

NewCDASender builds a sender for a cda destination.

func (*CDASender) Close 

func (s *CDASender) Close() error

Close releases open files.

func (*CDASender) Describe 

func (s *CDASender) Describe() string

Describe identifies the destination in logs.

func (*CDASender) Send 

func (s *CDASender) Send(ctx context.Context, raw []byte) error

Send extracts, converts and delivers.

func (*CDASender) Stats 

func (s *CDASender) Stats() CDAStats

Stats returns a snapshot.

type CDAStats 

type CDAStats struct {
	Extracted    int64 `json:"extracted"`
	NoDocument   int64 `json:"noDocument"`
	NotXML       int64 `json:"notXml"`
	Converted    int64 `json:"converted"`
	ConvertFails int64 `json:"conversionFailed"`
	Disagreed    int64 `json:"disagreed"`
	Sent         int64 `json:"sent"`
}

CDAStats reports what this sender has done.

type Channel 

type Channel struct {
	// contains filtered or unexported fields
}

Channel is a running channel.

func NewChannel 

func NewChannel(cfg *config.Channel, factory SenderFactory, log *slog.Logger) (*Channel, error)

NewChannel prepares a channel to run. Nothing is listening until Start.

func (*Channel) Close 

func (c *Channel) Close() error

Close releases the senders without needing a listener to have been started.

Stop is the normal path and it shuts down the server first. A test never starts one, so calling Stop would return an error about a server that was never there.

func (*Channel) HandleForTest 

func (c *Channel) HandleForTest(ctx context.Context, raw []byte) ([]byte, error)

HandleForTest processes one message and returns the acknowledgement, without a listener.

This is the same function the MLLP server calls. A test therefore exercises the filter, the declarative transformations and the script in the real order, with only the transport replaced.

func (*Channel) LastOutcomeForTest 

func (c *Channel) LastOutcomeForTest() Outcome

LastOutcomeForTest reports what happened to the most recent message.

func (*Channel) Name 

func (c *Channel) Name() string

Name returns the channel name.

func (*Channel) Route 

func (c *Channel) Route(ctx context.Context, raw []byte) ([]byte, error)

Route handles a message that arrived from another channel rather than from a source.

The same handling a received message gets, which is the point: a routed message must be filtered, transformed, recorded and delivered exactly as one that arrived over the wire, or the child channel would behave differently depending on where its traffic came from and nobody could reason about it.

The acknowledgement is returned for completeness. The usual caller discards it, because it is addressed to whoever sent the message and the sender is another channel.

func (*Channel) SetAdmission added in v0.1.11 

func (c *Channel) SetAdmission(intake func() error)

SetAdmission sets what is asked before each received message is processed. It must be called before Start.

func (*Channel) SetAdmit 

func (c *Channel) SetAdmit(a *admit.Controller)

SetAdmit gives this channel the process-wide delivery limiter.

Must be the same controller every channel gets, or each will believe it has the whole descriptor budget to itself and the limit will be the number of channels multiplied by what the process can afford. A channel without one is unlimited, which is right for a test and wrong for a server.

func (*Channel) SetMetrics 

func (c *Channel) SetMetrics(m *metrics.Collector)

SetMetrics attaches a collector. Passing nil disables recording, which is what tests and one-shot command-line runs do.

func (*Channel) SetQueryState 

func (c *Channel) SetQueryState(s DICOMQueryState)

SetQueryState gives a channel somewhere to record what its DICOM query source has already seen.

Set from outside rather than constructed here, because the state lives in the same database as the queue and the engine does not otherwise know about a database at all.

func (*Channel) SetQueues 

func (c *Channel) SetQueues(q *Queues)

SetQueues attaches a queue coordinator to the channel.

func (*Channel) SetRecorder 

func (c *Channel) SetRecorder(r Recorder)

SetRecorder attaches a recorder. It must be called before Start.

func (*Channel) SetTenant 

func (c *Channel) SetTenant(id string)

SetTenant records which tenant this channel belongs to, for metric labels.

Empty means single-tenant, which reports as the default tenant rather than as an empty label. An empty label value in Prometheus is indistinguishable from an absent one and would make a single-tenant series impossible to select.

func (*Channel) SetTracer 

func (c *Channel) SetTracer(t *trace.Tracer)

SetTracer attaches a tracer. A nil tracer is valid and disables tracing, which is why nothing on the message path guards against it.

Set after construction rather than passed in, matching SetMetrics: tracing is an operational concern that every existing caller and test should not have to mention.

func (*Channel) ShadowReport 

func (c *Channel) ShadowReport() *ShadowReport

ShadowReport returns the comparison so far, or nil if there is none.

func (*Channel) Start 

func (c *Channel) Start() error

Start begins listening. It returns once the listener is up.

func (*Channel) Stats 

func (c *Channel) Stats() Stats

Stats returns a snapshot of the counters.

func (*Channel) Stop 

func (c *Channel) Stop(ctx context.Context) error

Stop shuts the channel down, waiting for messages in flight.

func (*Channel) TransformOnly 

func (c *Channel) TransformOnly(ctx context.Context, raw []byte) (shadow.Result, error)

TransformOnly runs the filter and the transformation pipeline and returns the result without delivering anything.

This is what a shadow channel is given, and it is the only method it is given. A shadow holds a *Channel whose senders were never constructed, so there is nothing to deliver to even if something tried - but the narrow interface is what makes "the shadow cannot deliver" a fact about the code rather than a promise about it.

type ChannelRouter 

type ChannelRouter interface {
	// Route hands raw to the named channel and reports what happened to it.
	//
	// The error describes a delivery failure, not a filtered message: a channel that examines a
	// message and decides not to forward it has done its job.
	Route(ctx context.Context, channel string, raw []byte) error
}

ChannelRouter resolves a channel name and hands it a message.

An interface rather than a concrete type because the thing that owns running channels lives above this package, and the engine must not import it. Whoever builds the channels supplies the router.

type ChannelSender 

type ChannelSender struct {
	// contains filtered or unexported fields
}

ChannelSender hands messages to another channel in the same process.

func NewChannelSender 

func NewChannelSender(d config.Destination, router ChannelRouter) (*ChannelSender, error)

NewChannelSender builds a sender that routes to another channel.

func (*ChannelSender) Close 

func (s *ChannelSender) Close() error

Close releases nothing.

The target channel's lifetime belongs to whatever started it. Closing it here would stop a channel that other channels may also be routing to, and that other people may be watching.

func (*ChannelSender) Describe 

func (s *ChannelSender) Describe() string

Describe says where this destination sends.

func (*ChannelSender) Send 

func (s *ChannelSender) Send(ctx context.Context, raw []byte) error

Send hands the message to the target channel.

func (*ChannelSender) Target 

func (s *ChannelSender) Target() string

Target reports which channel this sender routes to, so loop detection can see the graph.

type DICOMQueryState 

type DICOMQueryState interface {
	Load(ctx context.Context, tenantID, channel string) (dicomstate.Mark, error)
	SaveMark(ctx context.Context, tenantID, channel string, highWater, polledAt time.Time) error
	Unseen(ctx context.Context, tenantID, channel string, identifiers []string) ([]string, error)
	MarkSeen(ctx context.Context, tenantID, channel string, identifiers []string, at time.Time) error
	Prune(ctx context.Context, tenantID, channel string, retain time.Duration, now time.Time) (int64, error)
}

DICOMQueryState is what a query source needs to remember between polls.

An interface so the channel can run without one - a test harness, a dry run - and so the engine does not depend on the database package directly.

type DICOMSender 

type DICOMSender struct {
	// contains filtered or unexported fields
}

DICOMSender stores objects to a remote archive.

func NewDICOMSender 

func NewDICOMSender(d config.Destination) (*DICOMSender, error)

NewDICOMSender builds the sender.

func (*DICOMSender) Close 

func (s *DICOMSender) Close() error

Close releases nothing: an association is opened per operation, deliberately, because a half-broken association is worse than none and a fresh one cannot be poisoned by the last failure.

func (*DICOMSender) Describe 

func (s *DICOMSender) Describe() string

Describe names the destination, never the credentials.

func (*DICOMSender) Send 

func (s *DICOMSender) Send(ctx context.Context, msg []byte) error

Send stores one object.

type DICOMwebSender added in v0.1.16 

type DICOMwebSender struct {
	// contains filtered or unexported fields
}

DICOMwebSender stores each object on a DICOMweb server with STOW-RS.

func NewDICOMwebSender added in v0.1.16 

func NewDICOMwebSender(d config.Destination) (*DICOMwebSender, error)

NewDICOMwebSender builds the sender.

func (*DICOMwebSender) Close added in v0.1.16 

func (s *DICOMwebSender) Close() error

Close releases nothing.

func (*DICOMwebSender) Describe added in v0.1.16 

func (s *DICOMwebSender) Describe() string

Describe names the destination, never the credentials.

func (*DICOMwebSender) Send added in v0.1.16 

func (s *DICOMwebSender) Send(ctx context.Context, msg []byte) error

Send stores one object. The server's per-object answer decides: a 200 listing the object as failed is a failure.

type DataTypeAware 

type DataTypeAware interface {
	SetDataType(config.DataType)
}

DataTypeAware is implemented by senders whose output depends on the message format.

An optional interface rather than a change to Sender, so the senders for which the format makes no difference - HTTP posts the bytes it is given, a database stores them - need no code at all. It follows the SetMetrics and SetRecorder pattern already used here: attached after construction, before anything is sent.

type DatabaseSender 

type DatabaseSender struct {
	// contains filtered or unexported fields
}

DatabaseSender writes each message to a database.

The whole of the interesting part is that values are bound as parameters and never built into the SQL. That is usually framed as an injection defence, and it is one, but the everyday reason is duller and hits sooner: O'Brien is a common name, and a query assembled by concatenation breaks on the apostrophe. A feed that works for a month and then fails for one patient is harder to diagnose than one that never worked.

func NewDatabaseSender 

func NewDatabaseSender(d config.Destination, log *slog.Logger) (*DatabaseSender, error)

NewDatabaseSender opens the pool and prepares the parameter bindings.

func (*DatabaseSender) Close 

func (s *DatabaseSender) Close() error

Close releases the pool.

func (*DatabaseSender) Describe 

func (s *DatabaseSender) Describe() string

Describe names the destination for the interface and the log.

The DSN is redacted, because this string appears on the channel page and in startup lines, and a DSN normally carries a password.

func (*DatabaseSender) Send 

func (s *DatabaseSender) Send(ctx context.Context, raw []byte) error

Send writes one message.

func (*DatabaseSender) Stats 

func (s *DatabaseSender) Stats() DatabaseStats

Stats returns a copy of the counters.

type DatabaseSourceStats 

type DatabaseSourceStats struct {
	Polls        int64     `json:"polls"`
	RowsRead     int64     `json:"rowsRead"`
	Accepted     int64     `json:"accepted"`
	Quarantined  int64     `json:"quarantined"`
	PollFailures int64     `json:"pollFailures"`
	LastPoll     time.Time `json:"lastPoll,omitempty"`
	LastError    string    `json:"lastError,omitempty"`
}

DatabaseSourceStats reports what the poller has done.

type DatabaseStats 

type DatabaseStats struct {
	Written   int64     `json:"written"`
	Failed    int64     `json:"failed"`
	NullBound int64     `json:"nullBound"`
	LastError string    `json:"lastError,omitempty"`
	LastWrite time.Time `json:"lastWrite,omitempty"`
}

DatabaseStats reports what the destination has done.

type DeliveryRecord 

type DeliveryRecord struct {
	Destination string
	// Status is "delivered", "failed" or "filtered".
	Status   string
	Attempts int
	Duration time.Duration
	Error    string
}

DeliveryRecord is the outcome at one destination.

type DirectSender added in v0.1.16 

type DirectSender struct {
	// contains filtered or unexported fields
}

DirectSender sends each message as a Direct secure message through the HISP.

func NewDirectSender added in v0.1.16 

func NewDirectSender(d config.Destination) (*DirectSender, error)

NewDirectSender builds the sender; the address's certificate and key are loaded now, so a bad path fails at deploy.

func (*DirectSender) Close added in v0.1.16 

func (s *DirectSender) Close() error

Close releases nothing.

func (*DirectSender) Describe added in v0.1.16 

func (s *DirectSender) Describe() string

Describe names the destination, never the credentials.

func (*DirectSender) Send added in v0.1.16 

func (s *DirectSender) Send(ctx context.Context, msg []byte) error

Send protects one message and submits it to the HISP.

func (*DirectSender) SetDataType added in v0.1.16 

func (s *DirectSender) SetDataType(t config.DataType)

SetDataType decides substitution and the attachment's default name.

type DocumentSender 

type DocumentSender struct {
	// contains filtered or unexported fields
}

DocumentSender renders each message into a document a person reads.

func NewDocumentSender 

func NewDocumentSender(d config.Destination) (*DocumentSender, error)

NewDocumentSender builds the sender.

func (*DocumentSender) Close 

func (s *DocumentSender) Close() error

Close releases nothing.

func (*DocumentSender) Describe 

func (s *DocumentSender) Describe() string

Describe names the destination.

func (*DocumentSender) Send 

func (s *DocumentSender) Send(_ context.Context, msg []byte) error

Send renders and writes one document.

type ECRStats added in v0.1.6 

type ECRStats struct {
	Reported      int64 `json:"reported"`
	NotReportable int64 `json:"notReportable"`
}

ECRStats counts what a case reporting destination did: reports sent, and messages it read and found nothing reportable in.

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine runs a set of channels.

func New 

func New(cfgs []*config.Channel, factory SenderFactory, log *slog.Logger) (*Engine, error)

New prepares an engine from loaded configuration.

Every channel is constructed before any is started, so a configuration error in the last file does not leave the first three listening.

func (*Engine) Channels 

func (e *Engine) Channels() []*Channel

Channels returns the prepared channels.

func (*Engine) Start 

func (e *Engine) Start() error

Start starts every channel. If one fails, those already started are stopped, so the process does not end up half up with nobody having said so.

func (*Engine) Stats 

func (e *Engine) Stats() map[string]Stats

Stats returns a snapshot per channel.

func (*Engine) Stop 

func (e *Engine) Stop(ctx context.Context) error

Stop shuts every channel down, waiting for messages in flight until ctx is done.

type FHIRSender 

type FHIRSender struct {
	// contains filtered or unexported fields
}

FHIRSender converts an HL7 v2 message to FHIR and posts it to a FHIR server.

This is the destination that makes the engine useful beyond v2: a hospital feed arrives as HL7 v2 and lands in a FHIR store, with no separate integration project in between.

It sends a transaction bundle rather than individual resources, so a patient and their encounter land together or not at all. Posting them separately leaves an encounter referring to a patient that does not exist if the second call fails.

func NewFHIRSender 

func NewFHIRSender(d config.Destination, log *slog.Logger) (*FHIRSender, error)

NewFHIRSender builds a sender from configuration.

func (*FHIRSender) Close 

func (s *FHIRSender) Close() error

Close releases idle connections.

func (*FHIRSender) Describe 

func (s *FHIRSender) Describe() string

Describe names the destination for logs.

func (*FHIRSender) ECRStats added in v0.1.6 

func (s *FHIRSender) ECRStats() *ECRStats

ECRStats is nil for an ordinary FHIR destination.

func (*FHIRSender) Send 

func (s *FHIRSender) Send(ctx context.Context, raw []byte) error

Send converts and posts one message.

func (*FHIRSender) SetBundleScript added in v0.1.21 

func (s *FHIRSender) SetBundleScript(engine *script.Engine, name, source string) error

SetBundleScript compiles the destination's bundle script on the channel's engine.

func (*FHIRSender) Stats 

func (s *FHIRSender) Stats() FHIRStats

Stats returns a snapshot.

type FHIRStats 

type FHIRStats struct {
	Converted        int64 `json:"converted"`
	ConversionFailed int64 `json:"conversionFailed"`
	Rejected         int64 `json:"rejected"`
	Sent             int64 `json:"sent"`
	Warnings         int64 `json:"warnings"`
}

FHIRStats reports what this sender has done.

type FTPSender 

type FTPSender struct {
	// contains filtered or unexported fields
}

FTPSender uploads a file per message.

A connection per message rather than a pooled one. FTP sessions are stateful - a current directory, a transfer mode, a half-finished rename - and a pooled session that failed part way through leaves the next message to inherit whatever state it was in. These servers are also frequently the kind that drop idle connections without saying so, which turns a pool into a source of intermittent failures on the first message after a quiet period. Reconnecting costs a few hundred milliseconds and buys a session that is provably clean.

func NewFTPSender 

func NewFTPSender(d config.Destination) (*FTPSender, error)

NewFTPSender builds the sender.

func (*FTPSender) Close 

func (s *FTPSender) Close() error

Close releases nothing, because nothing is held between messages.

func (*FTPSender) Describe 

func (s *FTPSender) Describe() string

Describe names the destination, never the password.

func (*FTPSender) Send 

func (s *FTPSender) Send(ctx context.Context, msg []byte) error

Send uploads one message.

func (*FTPSender) SetSeal added in v0.1.17 

func (s *FTPSender) SetSeal(f func([]byte) ([]byte, error))

SetSeal makes the sender encrypt each object's bytes (a pgp block).

type FileSender 

type FileSender struct {
	// contains filtered or unexported fields
}

FileSender appends messages to a file per day and message type.

One file per message would be tidier and produces millions of inodes on a busy feed, which is a problem people discover in production. Messages are stored MLLP-framed so a file replays byte for byte.

func NewFileSender 

func NewFileSender(d config.Destination) (*FileSender, error)

NewFileSender builds a sender for a file destination, creating the directory.

func (*FileSender) Close 

func (s *FileSender) Close() error

Close closes every open file.

func (*FileSender) Describe 

func (s *FileSender) Describe() string

Describe names the destination for logs.

func (*FileSender) Send 

func (s *FileSender) Send(ctx context.Context, msg []byte) error

Send appends the message.

func (*FileSender) SetDataType 

func (s *FileSender) SetDataType(t config.DataType)

SetDataType tells the sender what format it is writing.

Without it every archive was written as framed HL7. For an X12 channel that produced a file named .hl7, containing claims wrapped in MLLP control characters, that no X12 receiver would accept - a corrupted archive that looked like a successful delivery.

func (*FileSender) SetSeal added in v0.1.17 

func (s *FileSender) SetSeal(f func([]byte) ([]byte, error))

SetSeal makes the sender encrypt each file's bytes (a pgp block).

type FileSourceStats 

type FileSourceStats struct {
	Polls        int64     `json:"polls"`
	FilesRead    int64     `json:"filesRead"`
	FilesWaiting int64     `json:"filesWaiting"`
	FilesFailed  int64     `json:"filesFailed"`
	FilesSkipped int64     `json:"filesSkipped"`
	Messages     int64     `json:"messages"`
	PollFailures int64     `json:"pollFailures"`
	Backlog      int64     `json:"backlog"`
	LastPoll     time.Time `json:"lastPoll,omitempty"`
	LastError    string    `json:"lastError,omitempty"`
	Connected    bool      `json:"connected"`
}

FileSourceStats reports what a poller has done.

type GCSSender added in v0.1.10 

type GCSSender struct {
	// contains filtered or unexported fields
}

GCSSender writes each message as an object.

func NewGCSSender added in v0.1.10 

func NewGCSSender(d config.Destination) (*GCSSender, error)

NewGCSSender makes the sender.

func (*GCSSender) Close added in v0.1.10 

func (s *GCSSender) Close() error

Close does nothing.

func (*GCSSender) Describe added in v0.1.10 

func (s *GCSSender) Describe() string

Describe names the bucket.

func (*GCSSender) Send added in v0.1.10 

func (s *GCSSender) Send(ctx context.Context, msg []byte) error

Send writes the object.

func (*GCSSender) SetSeal added in v0.1.17 

func (s *GCSSender) SetSeal(f func([]byte) ([]byte, error))

SetSeal makes the sender encrypt each object's bytes (a pgp block).

type HTTPSender 

type HTTPSender struct {
	// contains filtered or unexported fields
}

HTTPSender posts messages to an HTTP endpoint.

Most of the care here is about what counts as delivered. An HTTP transport that treats any response as success is worse than no transport at all: the message is gone, the channel reports a delivery, and nobody finds out until somebody asks where a week of results went.

func NewHTTPSender 

func NewHTTPSender(d config.Destination, log *slog.Logger) (*HTTPSender, error)

NewHTTPSender builds a sender for an http destination.

func (*HTTPSender) Close 

func (s *HTTPSender) Close() error

Close releases idle connections.

func (*HTTPSender) Describe 

func (s *HTTPSender) Describe() string

Describe identifies the destination in logs.

func (*HTTPSender) Send 

func (s *HTTPSender) Send(ctx context.Context, raw []byte) error

Send posts one message.

func (*HTTPSender) Stats 

func (s *HTTPSender) Stats() HTTPStats

Stats returns a snapshot.

type HTTPStats 

type HTTPStats struct {
	Sent      int64 `json:"sent"`
	Rejected  int64 `json:"rejected"`
	BodyFails int64 `json:"bodyFailures"`
}

HTTPStats reports what this sender has done.

type IBMMQSender added in v0.1.10 

type IBMMQSender struct {
	// contains filtered or unexported fields
}

IBMMQSender puts each message on a queue.

func NewIBMMQSender added in v0.1.10 

func NewIBMMQSender(d config.Destination) (*IBMMQSender, error)

NewIBMMQSender makes the sender.

func (*IBMMQSender) Close added in v0.1.10 

func (s *IBMMQSender) Close() error

Close does nothing; each put is its own request.

func (*IBMMQSender) Describe added in v0.1.10 

func (s *IBMMQSender) Describe() string

Describe names the queue.

func (*IBMMQSender) Send added in v0.1.10 

func (s *IBMMQSender) Send(ctx context.Context, msg []byte) error

Send puts the message, with its HL7 control id as the correlation id when it has one.

type JSReaderStats 

type JSReaderStats struct {
	Polls        int64     `json:"polls"`
	Messages     int64     `json:"messages"`
	Errors       int64     `json:"errors"`
	LastPoll     time.Time `json:"lastPoll,omitempty"`
	LastError    string    `json:"lastError,omitempty"`
	LastDuration string    `json:"lastDuration,omitempty"`
}

JSReaderStats reports what the reader has done.

type JavaScriptSender 

type JavaScriptSender struct {
	// contains filtered or unexported fields
}

JavaScriptSender runs a script instead of sending anywhere.

Mirth's JavaScript Writer, and used far more than a feature list suggests: it is the escape hatch sites reach for when they need something no connector covers. A migration that cannot run those cannot move the channels containing them, which in a mature installation is a large fraction of them.

Constructed by the channel rather than the sender factory, because it needs the channel's script engine - so the shared maps a transformer wrote are the same ones this reads, which is what makes a script written for Mirth behave the same way here.

func (*JavaScriptSender) Close 

func (s *JavaScriptSender) Close() error

Close releases nothing.

func (*JavaScriptSender) Describe 

func (s *JavaScriptSender) Describe() string

Describe names the destination.

func (*JavaScriptSender) Send 

func (s *JavaScriptSender) Send(ctx context.Context, msg []byte) error

Send runs the script for one message.

type KafkaSender added in v0.1.1 

type KafkaSender struct {
	// contains filtered or unexported fields
}

KafkaSender publishes messages to a Kafka topic.

The connection is opened on demand and kept. A producer that reconnected per message would spend more time in metadata requests and handshakes than in producing, and a cluster sees a connection per message as an attack.

func NewKafkaSender added in v0.1.1 

func NewKafkaSender(d config.Destination) (*KafkaSender, error)

NewKafkaSender builds a sender from a destination.

Refused here rather than defaulted when the section is missing, matching the other senders: a destination that cannot work is refused at load rather than at the first patient message.

func (*KafkaSender) Close added in v0.1.1 

func (s *KafkaSender) Close() error

Close releases the connection.

func (*KafkaSender) Describe added in v0.1.1 

func (s *KafkaSender) Describe() string

Describe names this destination in logs and errors.

func (*KafkaSender) Send added in v0.1.1 

func (s *KafkaSender) Send(ctx context.Context, msg []byte) error

Send publishes one message.

type MLLPSender 

type MLLPSender struct {
	// contains filtered or unexported fields
}

MLLPSender forwards messages to a remote MLLP listener and treats a negative acknowledgement as a delivery failure.

func NewMLLPSender 

func NewMLLPSender(d config.Destination) (*MLLPSender, error)

NewMLLPSender builds a sender for an mllp destination.

func (*MLLPSender) Close 

func (s *MLLPSender) Close() error

Close releases the connection.

func (*MLLPSender) Describe 

func (s *MLLPSender) Describe() string

Describe names the destination for logs.

func (*MLLPSender) Send 

func (s *MLLPSender) Send(ctx context.Context, msg []byte) error

Send forwards the message and waits for the acknowledgement.

An AE or AR is a failure, not a success. Forwarding a message, being told it was rejected, and then reporting delivery upstream would lose the data with every party believing someone else has it.

func (*MLLPSender) SendForResponse 

func (s *MLLPSender) SendForResponse(ctx context.Context, msg []byte) ([]byte, error)

SendForResponse delivers the message and returns the acknowledgement, satisfying Responder.

The acknowledgement is returned even when the error is non-nil, because it is usually what explains the error - a negative acknowledgement carries the reason in MSA-3, and a response transformer that could not see it would be inspecting nothing on precisely the deliveries worth inspecting.

type MPISender added in v0.1.17 

type MPISender struct {
	// contains filtered or unexported fields
}

MPISender registers each message's patient in the patient index.

func NewMPISender added in v0.1.17 

func NewMPISender(d config.Destination) (*MPISender, error)

NewMPISender builds the sender. The index itself is the server's, so a channel can be loaded where there is none; it fails on the first message instead, saying why.

func (*MPISender) Close added in v0.1.17 

func (s *MPISender) Close() error

func (*MPISender) Describe added in v0.1.17 

func (s *MPISender) Describe() string

func (*MPISender) Send added in v0.1.17 

func (s *MPISender) Send(ctx context.Context, msg []byte) error

Send registers the patient, and applies a merge the message asks for.

func (*MPISender) SetChannel added in v0.1.17 

func (s *MPISender) SetChannel(name string)

SetChannel names the channel the records come from, which is what the index shows as their source.

func (*MPISender) SetTenant added in v0.1.17 

func (s *MPISender) SetTenant(id string)

SetTenant scopes the registrations to the channel's tenant.

type MQTTSender added in v0.1.10 

type MQTTSender struct {
	// contains filtered or unexported fields
}

MQTTSender publishes to an MQTT topic, keeping one connection open.

func NewMQTTSender added in v0.1.10 

func NewMQTTSender(d config.Destination) (*MQTTSender, error)

NewMQTTSender makes the sender; it connects on the first message.

func (*MQTTSender) Close added in v0.1.10 

func (s *MQTTSender) Close() error

Close disconnects.

func (*MQTTSender) Describe added in v0.1.10 

func (s *MQTTSender) Describe() string

Describe names the topic.

func (*MQTTSender) Send added in v0.1.10 

func (s *MQTTSender) Send(ctx context.Context, msg []byte) error

Send publishes; at QoS 1 it returns once the broker has the message.

type MessageRecord 

type MessageRecord struct {
	Channel      string
	ReceivedAt   time.Time
	Raw          []byte
	ControlID    string
	MessageType  string
	TriggerEvent string
	Sender       string
	Segments     int

	// Attachments are payloads extracted from Raw and replaced with tokens, so the recorder can store them once by
	// content rather than inline in every message that carries them.
	//
	// Carried on the record rather than written by the channel, because the channel has no database handle - and
	// giving it one would put storage on the acknowledgement path, which is exactly what the recorder exists to
	// avoid.
	Attachments []attach.Attachment

	// Transformed reports whether the message was changed on the way through, and
	// Changes how many declarative steps took effect. Both are recorded because
	// the most expensive failure in an interface is a transformation nobody knew
	// was happening.
	Transformed bool
	Changes     int
	Outcome     Outcome
	AckCode     string
	Duration    time.Duration
	Error       string
	Deliveries  []DeliveryRecord
	// contains filtered or unexported fields
}

MessageRecord is what a channel reports about one message.

type Outcome 

type Outcome string

Outcome is what happened to one received message. 

const (
	// Delivered means every destination that wanted the message accepted it.
	Delivered Outcome = "delivered"
	// Filtered means the channel filter rejected the message. Nothing was sent
	// anywhere, and the sender is still told AA: it did nothing wrong.
	Filtered Outcome = "filtered"
	// PartiallyDelivered means at least one destination accepted and at least
	// one failed.
	PartiallyDelivered Outcome = "partial"
	// Failed means no destination accepted the message.
	Failed Outcome = "failed"
	// Unparseable means the bytes were not an HL7 message.
	Unparseable Outcome = "unparseable"
	// Queued means the message could not be delivered now but is on disk and will
	// be retried. The sender is told AA, because the bytes are committed before the
	// acknowledgement goes out and we have therefore accepted responsibility.
	// Reporting an error instead would make a working store-and-forward queue look
	// like a fault and invite the sender to resend what we already hold.
	Queued Outcome = "queued"
)

Message outcomes. These are separate on purpose: "we chose not to take it" and "we could not take it" look the same to a sender that only sees an AA, and telling them apart is the difference between a quiet afternoon and a lost afternoon.

type PASSender added in v0.1.23 

type PASSender struct {
	// contains filtered or unexported fields
}

PASSender submits each message, a Da Vinci PAS request Bundle, to a payer's Claim/$submit: the provider's side of CMS-0057 prior authorization, for systems that build the request but do not send it.

A decision is the outcome, whatever it is: a denial or a pend is a delivered request, recorded with the payer's reference, not a failure to retry. What fails delivery is a payer that did not decide: an error status, an OperationOutcome, or a ClaimResponse whose outcome is error. The payer's whole answer is this destination's reply, for a response transformer, and goes to the channel results_to names, when there is one.

func NewPASSender added in v0.1.23 

func NewPASSender(d config.Destination, router ChannelRouter) (*PASSender, error)

NewPASSender builds the sender.

func (*PASSender) Close added in v0.1.23 

func (s *PASSender) Close() error

Close releases nothing.

func (*PASSender) Describe added in v0.1.23 

func (s *PASSender) Describe() string

Describe says where requests go.

func (*PASSender) Send added in v0.1.23 

func (s *PASSender) Send(ctx context.Context, msg []byte) error

Send submits the request.

func (*PASSender) SendForResponse added in v0.1.23 

func (s *PASSender) SendForResponse(ctx context.Context, msg []byte) ([]byte, error)

SendForResponse submits the request and returns the payer's answer.

type PubSubSender added in v0.1.10 

type PubSubSender struct {
	// contains filtered or unexported fields
}

PubSubSender publishes to a topic.

func NewPubSubSender added in v0.1.10 

func NewPubSubSender(d config.Destination) (*PubSubSender, error)

NewPubSubSender makes the sender.

func (*PubSubSender) Close added in v0.1.10 

func (s *PubSubSender) Close() error

Close does nothing; each publish is its own request.

func (*PubSubSender) Describe added in v0.1.10 

func (s *PubSubSender) Describe() string

Describe names the topic.

func (*PubSubSender) Send added in v0.1.10 

func (s *PubSubSender) Send(ctx context.Context, msg []byte) error

Send publishes, with the message type and control id as attributes when it is HL7 v2.

type Queues 

type Queues struct {
	// contains filtered or unexported fields
}

Queues wires a channel to a durable queue. Nil means queueing is off.

func NewQueues 

func NewQueues(store *queue.Store, log *slog.Logger) *Queues

NewQueues builds the coordinator.

func (*Queues) Stop 

func (q *Queues) Stop()

Stop halts every worker and waits for them.

A worker mid-delivery is allowed to finish that attempt, because abandoning it would leave a message we may already have sent marked as pending, and the retry would duplicate it.

func (*Queues) Store 

func (q *Queues) Store() *queue.Store

Store exposes the queue for the API.

func (*Queues) WakeAll 

func (q *Queues) WakeAll()

WakeAll asks every worker to look now.

Called after an operator presses retry: somebody who has just fixed a receiver should see the queue move, not wait out a poll interval wondering whether the button worked.

type Recorder 

type Recorder interface {
	RecordMessage(ctx context.Context, record MessageRecord)
}

Recorder receives a record of every message a channel handled.

The engine does not know about storage: it hands over what happened and lets something else decide whether to keep it. That keeps the message path free of database concerns and means a deployment that must not store clinical content simply supplies no recorder.

type ReplayDifference 

type ReplayDifference struct {
	// ControlID and MessageType identify the message so it can be found again in the store.
	ControlID   string `json:"controlId"`
	MessageType string `json:"messageType"`

	// Fields are the paths that differ. Empty with Verdict set means the disagreement is
	// about whether to accept the message at all, which no field diff can express.
	Fields []shadow.FieldDifference `json:"fields,omitempty"`

	// Verdict describes a difference in acceptance rather than content.
	Verdict string `json:"verdict,omitempty"`
}

ReplayDifference is one message that the two versions disagree about.

type ReplayReport 

type ReplayReport struct {
	// Examined is how many stored messages were run through both versions.
	Examined int `json:"examined"`
	// Identical is how many came out the same.
	Identical int `json:"identical"`
	// Changed is how many differ. This is the number that matters.
	Changed int `json:"changed"`
	// Unparseable counts stored messages neither version could read. Reported rather than
	// hidden: a store full of them means the sample is not representative, and a run that
	// silently skipped them would look reassuring for the wrong reason.
	Unparseable int `json:"unparseable"`

	// Differences are examples, bounded. A candidate that rewrites every message would
	// otherwise produce a report too long to read, which buries the one case somebody
	// needed to see.
	Differences []ReplayDifference `json:"differences"`
	// Truncated says the example list was cut short.
	Truncated bool `json:"truncated"`

	// FieldCounts is how many messages each differing path accounts for. This is usually
	// the most useful part: one path against every message is a deliberate change, and one
	// path against three messages out of five thousand is the edge case worth looking at.
	FieldCounts map[string]int `json:"fieldCounts,omitempty"`
}

ReplayReport is what a retroactive run produces.

func Replay 

func Replay(
	ctx context.Context,
	current, candidate *config.Channel,
	messages [][]byte,
	opts shadow.DiffOptions,
	log *slog.Logger,
) (*ReplayReport, error)

Replay runs stored messages through two versions of a channel and reports where they differ.

Neither version can deliver anything. Both are built with newShadowChannel, so there is no sender object in existence for either - "this cannot reach a receiver" is a fact about what is in memory rather than a promise about what a function does. That property is what makes it safe to run a candidate against real production traffic.

type Responder 

type Responder interface {
	// SendForResponse delivers the message and returns the receiver's reply.
	//
	// The reply is returned even when the error is non-nil, because the reply is usually what explains
	// the error - a negative acknowledgement carries the reason in its text.
	SendForResponse(ctx context.Context, msg []byte) ([]byte, error)
}

Responder is a Sender that can report what the receiver said.

type S3Sender 

type S3Sender struct {
	// contains filtered or unexported fields
}

S3Sender writes each message as an object.

func NewS3Sender 

func NewS3Sender(d config.Destination) (*S3Sender, error)

NewS3Sender builds the sender.

func (*S3Sender) Close 

func (s *S3Sender) Close() error

Close releases nothing: the HTTP client is shared and has no state to release.

func (*S3Sender) Describe 

func (s *S3Sender) Describe() string

Describe names the destination for logs and specifications.

The bucket and key pattern, never the credentials. A description ends up in a specification document that gets emailed to a vendor.

func (*S3Sender) Send 

func (s *S3Sender) Send(ctx context.Context, msg []byte) error

Send writes one object.

func (*S3Sender) SetSeal added in v0.1.17 

func (s *S3Sender) SetSeal(f func([]byte) ([]byte, error))

SetSeal makes the sender encrypt each object's bytes (a pgp block).

type SFTPSender 

type SFTPSender struct {
	// contains filtered or unexported fields
}

SFTPSender writes each message to a file on an SFTP server.

The design decision that matters is the temporary name. A file appearing at its final name while it is still being written is indistinguishable, to whoever collects it, from a finished file - which is the same problem this engine has reading files, and it is only polite to not inflict it. So every file is written under a suffix and renamed once it is closed, because a rename within a directory is atomic on every server worth using.

func NewSFTPSender 

func NewSFTPSender(d config.Destination, log *slog.Logger) (*SFTPSender, error)

NewSFTPSender prepares the destination.

The connection is not opened here. An SFTP server being briefly unreachable is completely normal, and refusing to start a channel because of it would turn a transient network problem into an outage needing a person.

func (*SFTPSender) Close 

func (s *SFTPSender) Close() error

Close ends the connection.

func (*SFTPSender) Describe 

func (s *SFTPSender) Describe() string

Describe names the destination for the interface and the log.

func (*SFTPSender) Send 

func (s *SFTPSender) Send(ctx context.Context, raw []byte) error

Send writes one message.

func (*SFTPSender) SetSeal added in v0.1.17 

func (s *SFTPSender) SetSeal(f func([]byte) ([]byte, error))

SetSeal makes the sender encrypt each object's bytes (a pgp block).

func (*SFTPSender) Stats 

func (s *SFTPSender) Stats() SFTPStats

Stats returns a copy of the counters.

type SFTPSourceStats 

type SFTPSourceStats struct {
	Polls        int64     `json:"polls"`
	FilesRead    int64     `json:"filesRead"`
	FilesWaiting int64     `json:"filesWaiting"`
	FilesFailed  int64     `json:"filesFailed"`
	Messages     int64     `json:"messages"`
	PollFailures int64     `json:"pollFailures"`
	LastPoll     time.Time `json:"lastPoll,omitempty"`
	LastError    string    `json:"lastError,omitempty"`
	Connected    bool      `json:"connected"`
}

SFTPSourceStats reports what the poller has done.

type SFTPStats 

type SFTPStats struct {
	Written     int64     `json:"written"`
	Failed      int64     `json:"failed"`
	Reconnects  int64     `json:"reconnects"`
	BytesGone   int64     `json:"bytesWritten"`
	LastWrite   time.Time `json:"lastWrite,omitempty"`
	LastError   string    `json:"lastError,omitempty"`
	Connected   bool      `json:"connected"`
	CurrentFile string    `json:"currentFile,omitempty"`
}

SFTPStats reports what the destination has done.

type SMTPSender 

type SMTPSender struct {
	// contains filtered or unexported fields
}

SMTPSender delivers by email.

func NewSMTPSender 

func NewSMTPSender(d config.Destination) (*SMTPSender, error)

NewSMTPSender builds an email sender.

func (*SMTPSender) Close 

func (s *SMTPSender) Close() error

Close releases resources. There are none: a connection is opened per message and closed when it is sent, because a mail server will drop an idle session anyway and a stale one fails on the next delivery rather than at the point it was lost.

func (*SMTPSender) Describe 

func (s *SMTPSender) Describe() string

func (*SMTPSender) Send 

func (s *SMTPSender) Send(ctx context.Context, raw []byte) error

Send delivers one message.

func (*SMTPSender) SetDataType 

func (s *SMTPSender) SetDataType(t config.DataType)

SetDataType records the format, which names the attachment.

type SNSSender added in v0.1.5 

type SNSSender struct {
	// contains filtered or unexported fields
}

SNSSender publishes each message to a topic.

func NewSNSSender added in v0.1.5 

func NewSNSSender(d config.Destination) (*SNSSender, error)

NewSNSSender builds the sender.

func (*SNSSender) Close added in v0.1.5 

func (s *SNSSender) Close() error

Close releases nothing.

func (*SNSSender) Describe added in v0.1.5 

func (s *SNSSender) Describe() string

Describe names the topic.

func (*SNSSender) Send added in v0.1.5 

func (s *SNSSender) Send(ctx context.Context, msg []byte) error

Send publishes one message.

type SOAPSender 

type SOAPSender struct {
	// contains filtered or unexported fields
}

SOAPSender posts a SOAP request per message.

func NewSOAPSender 

func NewSOAPSender(d config.Destination) (*SOAPSender, error)

NewSOAPSender builds the sender.

func (*SOAPSender) Close 

func (s *SOAPSender) Close() error

Close releases nothing that is not shared.

func (*SOAPSender) ConnectivityTest added in v0.1.16 

func (s *SOAPSender) ConnectivityTest(ctx context.Context, echo string) (string, error)

ConnectivityTest calls the registry's connectivityTest, which proves the address, TLS and SOAP binding without submitting anything. Credentials are not checked by this operation, so it cannot prove those.

func (*SOAPSender) Describe 

func (s *SOAPSender) Describe() string

Describe names the destination, never the password.

func (*SOAPSender) Send 

func (s *SOAPSender) Send(ctx context.Context, msg []byte) error

Send posts one request.

func (*SOAPSender) SendForResponse 

func (s *SOAPSender) SendForResponse(ctx context.Context, msg []byte) ([]byte, error)

SendForResponse posts one request and returns the response body, satisfying Responder.

Implemented so a response transformer can inspect the reply, which matters more for SOAP than for anything else here: these services routinely answer with a 200 and a body describing a rejection, and the fault codes that should count as success are entirely site-specific.

type SQSSender added in v0.1.5 

type SQSSender struct {
	// contains filtered or unexported fields
}

SQSSender sends each message to a queue.

func NewSQSSender added in v0.1.5 

func NewSQSSender(d config.Destination) (*SQSSender, error)

NewSQSSender builds the sender.

func (*SQSSender) Close added in v0.1.5 

func (s *SQSSender) Close() error

Close releases nothing.

func (*SQSSender) Describe added in v0.1.5 

func (s *SQSSender) Describe() string

Describe names the queue, never the credentials.

func (*SQSSender) Send added in v0.1.5 

func (s *SQSSender) Send(ctx context.Context, msg []byte) error

Send puts one message on the queue.

type Sender 

type Sender interface {
	Send(ctx context.Context, msg []byte) error
	// Describe is used in logs and errors.
	Describe() string
	// Close releases any resources.
	Close() error
}

Sender delivers a message to one destination.

An implementation must be safe for concurrent use and must respect the context deadline. Returning an error means the attempt failed and may be retried.

func DefaultSenderFactory 

func DefaultSenderFactory(d config.Destination) (Sender, error)

DefaultSenderFactory builds the sender for a configured destination.

type SenderFactory 

type SenderFactory func(config.Destination) (Sender, error)

SenderFactory builds a Sender for a destination. Injectable so the runtime can be tested without opening sockets.

func RoutingSenderFactory 

func RoutingSenderFactory(router ChannelRouter) SenderFactory

RoutingSenderFactory builds senders that can also route to other channels.

Everything except a channel destination is built exactly as the default factory builds it. Wrapping rather than duplicating means a new destination type is available to a routing server the moment it exists, with nothing to remember.

type SerialStats 

type SerialStats struct {
	Open          bool      `json:"open"`
	Messages      int64     `json:"messages"`
	FramingErrors int64     `json:"framingErrors"`
	Reopens       int64     `json:"reopens"`
	LastMessage   time.Time `json:"lastMessage,omitempty"`
	LastError     string    `json:"lastError,omitempty"`
	Framing       string    `json:"framing,omitempty"`

	// Quiet is true when nothing has arrived for longer than quiet_after.
	//
	// Reported as state rather than only logged, so the channel list can show it. A serial feed that has gone quiet looks
	// identical to one that is working until somebody asks.
	Quiet bool `json:"quiet"`
}

SerialStats reports what the reader has done.

type ShadowReport 

type ShadowReport struct {
	Channel     string              `json:"channel"`
	Candidate   string              `json:"candidate"`
	Stats       shadow.Stats        `json:"stats"`
	Verdict     string              `json:"verdict"`
	Differences []shadow.Difference `json:"differences"`
}

ShadowReport is what the interface shows.

type Stats 

type Stats struct {
	// Received is messages accepted from the source, and the unit is a message rather than a delivery from the sender's point of
	// view. That distinction only bites on delimited channels, where one document becomes one message per row: a four-hundred-row
	// file counts four hundred, not one.
	//
	// Recorded here because it was a consequence rather than a choice - rows were chosen so a filter could drop one bad row
	// instead of a whole file, and the counting followed - and because a site whose dashboard predates a delimited channel would
	// read the change as a hundredfold traffic increase. The header row is not counted; it is structure.
	Received    int64
	Delivered   int64
	Filtered    int64
	Partial     int64
	Failed      int64
	Unparseable int64
	// DatabaseQuarantined counts rows abandoned by a database source.
	DatabaseQuarantined int64
	// Queued counts messages accepted for later delivery. Separate from Delivered
	// because a dashboard that adds them together hides a backlog.
	Queued        int64
	DestDelivered map[string]int64
	DestFailed    map[string]int64
	DestFiltered  map[string]int64
	DestQueued    map[string]int64
}

Stats counts what a channel has done. Counters only, so reading them is cheap and they can be exposed without locking anything interesting.

type TCPSender 

type TCPSender struct {
	// contains filtered or unexported fields
}

TCPSender writes messages to a socket with configurable framing.

Mirth's TCP Sender. The counterpart to the TCP source, and the two share their framing code so that Perfuse sending to itself round-trips - which is not a trick, it is the only cheap way to be sure the two halves agree.

func NewTCPSender 

func NewTCPSender(cfg config.TCPDest) (*TCPSender, error)

NewTCPSender prepares a sender. It does not connect.

Deliberately not connecting here. A destination that dialled at construction would stop a whole channel from starting because one device was switched off overnight, and the queue exists precisely so that does not happen.

func (*TCPSender) Close 

func (s *TCPSender) Close() error

Close releases a held connection.

func (*TCPSender) Describe 

func (s *TCPSender) Describe() string

Describe names the destination for logs and the interface.

func (*TCPSender) Send 

func (s *TCPSender) Send(ctx context.Context, msg []byte) error

Send delivers one message.

type TCPSourceStats 

type TCPSourceStats struct {
	Accepted      int64  `json:"accepted"`
	Refused       int64  `json:"refused"`
	Open          int64  `json:"open"`
	Messages      int64  `json:"messages"`
	FramingErrors int64  `json:"framingErrors"`
	Oversize      int64  `json:"oversize"`
	LastError     string `json:"lastError,omitempty"`
	Framing       string `json:"framing,omitempty"`
}

TCPSourceStats reports what the listener has done.

type TEFCASender added in v0.1.23 

type TEFCASender struct {
	// contains filtered or unexported fields
}

TEFCASender asks the Facilitated FHIR partner for the records of each message's patient.

The patient is read as the patient index reads it: an HL7 v2 PID, or a FHIR Patient on a raw channel. The query goes by identifier when there is one, and by name with date of birth otherwise; a name alone is refused by the exchange, because a wrong match on a name does not fail, it returns somebody else's record. Every attempt is audited in the TEFCA trail, refused ones included.

The Bundle the partner returns is this destination's reply, so a response transformer can read it, and is handed to the channel results_to names, when there is one, as a message of its own.

func NewTEFCASender added in v0.1.23 

func NewTEFCASender(d config.Destination, router ChannelRouter) (*TEFCASender, error)

NewTEFCASender builds the sender. The participant is the server's, so a channel can load where there is none; it fails on the first message instead, saying why.

func (*TEFCASender) Close added in v0.1.23 

func (s *TEFCASender) Close() error

Close releases nothing; the exchange is the server's.

func (*TEFCASender) Describe added in v0.1.23 

func (s *TEFCASender) Describe() string

Describe says where this destination asks.

func (*TEFCASender) Send added in v0.1.23 

func (s *TEFCASender) Send(ctx context.Context, msg []byte) error

Send asks the partner and passes on what it returns.

func (*TEFCASender) SendForResponse added in v0.1.23 

func (s *TEFCASender) SendForResponse(ctx context.Context, msg []byte) ([]byte, error)

SendForResponse asks the partner and returns the Bundle it answered with.

type TenantRouter added in v0.1.23 

type TenantRouter interface {
	ChannelRouter
	TenantName() string
}

TenantRouter is a router that says which tenant it routes for, so what a sender records is kept to that tenant.

type Trace 

type Trace struct {
	Stages []TraceStage `json:"stages"`

	// Accepted says whether the message would be forwarded.
	Accepted bool `json:"accepted"`
	// Output is the message as it would leave, when it was accepted.
	Output string `json:"output,omitempty"`
	// Input is the message as it arrived, so the two can be compared side by side.
	Input string `json:"input"`

	// Destinations describes which destinations would receive it and which would not, and why.
	Destinations []TraceDestination `json:"destinations,omitempty"`

	// Caveat is set when the trace cannot describe the whole channel.
	Caveat string `json:"caveat,omitempty"`

	// Stale reports that the channel has been changed since the message arrived, so this trace describes what would
	// happen now rather than what happened then.
	//
	// Its own field as well as being in the caveat, so the interface can mark it rather than relying on somebody
	// reading a paragraph. A caveat in prose is a caveat somebody skims.
	Stale bool `json:"stale,omitempty"`
}

Trace is the whole journey of one message.

func TraceMessage 

func TraceMessage(ctx context.Context, cfg *config.Channel, raw []byte) (*Trace, error)

TraceMessage follows one message through a channel without delivering anything.

func TraceStoredMessage 

func TraceStoredMessage(ctx context.Context, cfg *config.Channel, raw []byte, arrived time.Time, changedAt time.Time) (*Trace, error)

TraceStoredMessage traces a message that has already been through the channel.

The distinction from TraceMessage is the whole reason this exists, and it was not stated anywhere: a trace runs the channel as it is configured now. For a message that arrived five minutes ago that is the same thing. For one that arrived last week, on a channel somebody has edited since, it is not - the trace explains what would happen to that message today, not what did happen to it then.

Nothing said so. Somebody investigating why a message was delivered, on a channel whose filter was tightened yesterday, would be shown a trace saying it would be filtered - and would reasonably conclude the delivery record was wrong.

So this takes the time the message arrived and says plainly whether the channel has changed since.

type TraceDestination 

type TraceDestination struct {
	Name string `json:"name"`
	// Would is true when this destination would receive the message.
	Would bool `json:"would"`
	// Why explains the decision.
	Why string `json:"why"`
	// Reads are the fields the destination filter looked at.
	Reads []TraceValue `json:"reads,omitempty"`
}

TraceDestination is one destination's decision about this message.

type TraceStage 

type TraceStage struct {
	// Stage names what ran: "parse", "filter", "transform", "output".
	Stage string `json:"stage"`
	// OK is false when this stage rejected or failed.
	OK bool `json:"ok"`
	// Detail explains the outcome in a sentence.
	Detail string `json:"detail"`

	// Reads are the fields this stage looked at, with their values. For a filter this is the
	// explanation: knowing the filter tested MSH-9.1 and found "ORU" answers the question
	// completely, without needing to reason about the expression.
	Reads []TraceValue `json:"reads,omitempty"`

	// Changes are what this stage altered.
	Changes []transform.Change `json:"changes,omitempty"`

	// Skipped are steps that ran and changed nothing, with why.
	//
	// The part of a trace people need most and the part no output had. A step addressing PID-19 on a feed that has
	// never sent PID-19 is not an error and produces no log line - it simply does nothing, for years.
	Skipped []transform.Skip `json:"skipped,omitempty"`

	// Steps breaks the transform stage into one entry per declarative step, each carrying the message as it stood
	// after that step ran.
	//
	// Only populated on the transform stage. Everything above says what the stage did as a whole, which answers
	// "what changed" but not "which of my fourteen steps changed it, and what did the message look like just
	// before". Those are the questions somebody has when a field is wrong and the pipeline is long.
	Steps []TraceStep `json:"steps,omitempty"`
}

TraceStage is one thing that happened to the message.

type TraceStep 

type TraceStep struct {
	// Number is the step's position in the pipeline, from 1, so it can be cited.
	Number int `json:"number"`

	// Label is the step's description, or what it does when the author wrote none.
	Label string `json:"label"`

	// Path is the field the step addresses.
	//
	// The most useful single thing about a step that did nothing: almost always the field is absent, and this is
	// what somebody compares against a real message.
	Path string `json:"path,omitempty"`

	// Outcome is "changed", "no-effect", "skipped" or "failed".
	Outcome string `json:"outcome"`

	// Detail explains the outcome in a sentence, including on success. "Ran" without saying what it did leaves the
	// reader to diff two blobs of pipe-delimited text.
	Detail string `json:"detail"`

	// Changes are what this step altered.
	Changes []transform.Change `json:"changes,omitempty"`

	// Message is the whole message as it stood after this step.
	//
	// Held per step rather than reconstructed from the changes, because the thing somebody wants to look at is the
	// message immediately before the step that broke it, and a list of deltas cannot show that without being
	// replayed to get there.
	Message string `json:"message"`
}

TraceStep is one declarative step, and the message immediately after it.

type TraceValue 

type TraceValue struct {
	Path  string `json:"path"`
	Value string `json:"value"`
	// Present distinguishes an empty field from an absent one, which is a distinction filters
	// turn on and people routinely conflate.
	Present bool `json:"present"`
}

TraceValue is one path and what was in it.

type XDSSender added in v0.1.16 

type XDSSender struct {
	// contains filtered or unexported fields
}

XDSSender publishes each clinical document to an XDS.b repository with ITI-41. The repository's RegistryResponse decides: a Failure, or an error inside a PartialSuccess, is a failed delivery with the registry's error codes.

func NewXDSSender added in v0.1.16 

func NewXDSSender(d config.Destination) (*XDSSender, error)

NewXDSSender builds the sender.

func (*XDSSender) Close added in v0.1.16 

func (s *XDSSender) Close() error

Close releases nothing.

func (*XDSSender) Describe added in v0.1.16 

func (s *XDSSender) Describe() string

Describe names the destination, never the credentials.

func (*XDSSender) Send added in v0.1.16 

func (s *XDSSender) Send(ctx context.Context, msg []byte) error

Send publishes one document.

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.