Architecting HL7 and FHIR Interoperability in Modern Hospital Information Systems

public sealed class MllpSocketListener : BackgroundService
{
    private const byte StartBlock = 0x0B;
    private const byte EndBlock = 0x1C;
    private const byte CarriageReturn = 0x0D;
    private readonly TcpListener _listener = new(IPAddress.Any, 2575);
    private readonly ILogger<MllpSocketListener> _logger;
    private readonly ChannelWriter<Hl7MessagePayload> _inboundQueue;

    public MllpSocketListener(
        ChannelWriter<Hl7MessagePayload> inboundQueue,
        ILogger<MllpSocketListener> logger)
    {
        _inboundQueue = inboundQueue;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _listener.Start();
        _logger.LogInformation("MLLP Listener running on port 2575");

        while (!stoppingToken.IsCancellationRequested)
        {
            var client = await _listener.AcceptTcpClientAsync(stoppingToken);
            _ = ProcessClientAsync(client, stoppingToken);
        }
    }

    private async Task ProcessClientAsync(TcpClient client, CancellationToken ct)
    {
        using (client)
        await using (var stream = client.GetStream())
        {
            var pipeReader = PipeReader.Create(stream);
            while (!ct.IsCancellationRequested)
            {
                var result = await pipeReader.ReadAsync(ct);
                var buffer = result.Buffer;

                while (TryExtractMllpMessage(ref buffer, out var hl7Raw))
                {
                    // Generate and write immediate ACK to satisfy hardware timeout
                    var ackBytes = GenerateHl7Ack(hl7Raw);
                    await stream.WriteAsync(ackBytes, ct);

                    // Enqueue raw payload for asynchronous FHIR transformation
                    await _inboundQueue.WriteAsync(new Hl7MessagePayload(hl7Raw), ct);
                }

                pipeReader.AdvanceTo(buffer.Start, buffer.End);
                if (result.IsCompleted) break;
            }
        }
    }
}

import { Observation, Coding } from 'fhir/r4';

interface ObxSegment {
  setId: string;
  valueType: 'NM' | 'ST' | 'TX'; // Numeric, String, Text
  identifier: string;            // e.g. "HGB^Hemoglobin^LOCAL"
  value: string;
  units: string;
  referenceRange: string;
  abnormalFlags: string;
  status: string;
}

export function mapObxToFhirObservation(
  patientId: string,
  encounterId: string,
  obx: ObxSegment,
  loincCode: string,
  loincDisplay: string
): Observation {
  return {
    resourceType: 'Observation',
    status: 'final',
    category: [
      {
        coding: [
          {
            system: 'http://terminology.hl7.org/CodeSystem/observation-category',
            code: 'laboratory',
            display: 'Laboratory',
          },
        ],
      },
    ],
    code: {
      coding: [
        {
          system: 'http://loinc.org',
          code: loincCode,
          display: loincDisplay,
        },
        {
          system: 'urn:dsf:analyzer:local',
          code: obx.identifier.split('^')[0],
          display: obx.identifier.split('^')[1] || 'Local Code',
        },
      ],
      text: loincDisplay,
    },
    subject: {
      reference: `Patient/${patientId}`,
    },
    encounter: {
      reference: `Encounter/${encounterId}`,
    },
    effectiveDateTime: new Date().toISOString(),
    valueQuantity: obx.valueType === 'NM' ? {
      value: parseFloat(obx.value),
      unit: obx.units,
      system: 'http://unitsofmeasure.org',
      code: obx.units,
    } : undefined,
    valueString: obx.valueType !== 'NM' ? obx.value : undefined,
    referenceRange: obx.referenceRange ? [
      {
        text: obx.referenceRange,
      },
    ] : undefined,
  };
}