Platform/angela/14116/read hl7

prime-router/src/main/kotlin/SettingsProvider.kt

@@ -5,7 +5,6 @@ import com.fasterxml.jackson.annotation.JsonValue
 import gov.cdc.prime.router.CustomerStatus.ACTIVE
 import gov.cdc.prime.router.CustomerStatus.INACTIVE
 import gov.cdc.prime.router.CustomerStatus.TESTING
-import gov.cdc.prime.router.fhirengine.utils.HL7Reader
 import gov.cdc.prime.router.validation.IItemValidator
 import gov.cdc.prime.router.validation.MarsOtcElrOnboardingValidator
 import gov.cdc.prime.router.validation.MarsOtcElrValidator
@@ -55,7 +54,6 @@ enum class Topic(
     val isUniversalPipeline: Boolean = true,
     val isSendOriginal: Boolean = false,
     val validator: IItemValidator = NoopItemValidator(),
-    val hl7ParseConfiguration: HL7Reader.Companion.HL7MessageParseAndConvertConfiguration? = null,
 ) {
     FULL_ELR("full-elr", true, false),
     ETOR_TI("etor-ti", true, false),

prime-router/src/main/kotlin/SubmissionReceiver.kt

@@ -9,8 +9,8 @@ import gov.cdc.prime.router.azure.ReportWriter
 import gov.cdc.prime.router.azure.WorkflowEngine
 import gov.cdc.prime.router.azure.db.enums.TaskAction
 import gov.cdc.prime.router.fhirengine.engine.FhirConvertQueueMessage
-import gov.cdc.prime.router.fhirengine.engine.MessageType
 import gov.cdc.prime.router.fhirengine.utils.FhirTranscoder
+import gov.cdc.prime.router.fhirengine.utils.HL7MessageHelpers
 import gov.cdc.prime.router.fhirengine.utils.HL7Reader
 
 /**
@@ -268,14 +268,14 @@ class UniversalPipelineReceiver : SubmissionReceiver {
 
         when (sender.format) {
             MimeFormat.HL7 -> {
-                val messages = HL7Reader(actionLogs).getMessages(content)
-                val isBatch = HL7Reader(actionLogs).isBatch(content, messages.size)
-                // create a Report for this incoming HL7 message to use for tracking in the database
+                val messageCount = HL7MessageHelpers.messageCount(content)
+                val isBatch = HL7Reader.isBatch(content, messageCount)
 
+                // create a Report for this incoming HL7 message to use for tracking in the database
                 report = Report(
                     if (isBatch) MimeFormat.HL7_BATCH else MimeFormat.HL7,
                     sources,
-                    messages.size,
+                    messageCount,
                     metadata = metadata,
                     nextAction = TaskAction.convert,
                     topic = sender.topic,
@@ -290,11 +290,8 @@ class UniversalPipelineReceiver : SubmissionReceiver {
 //                        actionLogs
 //                    )
 //                }
-
-                // check for valid message type
-                messages.forEachIndexed {
-                    idx, element ->
-                    MessageType.validateMessageType(element, actionLogs, idx + 1)
+                if (messageCount == 0 && !actionLogs.hasErrors()) {
+                    actionLogs.error(InvalidReportMessage("Unable to find HL7 messages in provided data."))
                 }
             }
 

prime-router/src/main/kotlin/azure/service/SubmissionResponseBuilder.kt

@@ -5,12 +5,12 @@ import com.google.common.net.HttpHeaders
 import com.microsoft.azure.functions.HttpRequestMessage
 import com.microsoft.azure.functions.HttpResponseMessage
 import com.microsoft.azure.functions.HttpStatus
-import gov.cdc.prime.router.ActionLogger
 import gov.cdc.prime.router.Sender
 import gov.cdc.prime.router.azure.HttpUtilities
 import gov.cdc.prime.router.azure.HttpUtilities.Companion.isSuccessful
 import gov.cdc.prime.router.common.JacksonMapperUtilities
 import gov.cdc.prime.router.fhirengine.translation.hl7.utils.HL7ACKUtils
+import gov.cdc.prime.router.fhirengine.utils.HL7MessageHelpers
 import gov.cdc.prime.router.fhirengine.utils.HL7Reader
 import gov.cdc.prime.router.history.DetailedSubmissionHistory
 import org.apache.logging.log4j.kotlin.Logging
@@ -106,12 +106,11 @@ class SubmissionResponseBuilder(
             contentType == HttpUtilities.hl7V2MediaType &&
             requestBody != null
         ) {
-            val hl7Reader = HL7Reader(ActionLogger())
-            val messages = hl7Reader.getMessages(requestBody)
-            val isBatch = hl7Reader.isBatch(requestBody, messages.size)
+            val messageCount = HL7MessageHelpers.messageCount(requestBody)
+            val isBatch = HL7Reader.isBatch(requestBody, messageCount)
 
-            if (!isBatch && messages.size == 1) {
-                val message = messages.first()
+            if (!isBatch && messageCount == 1) {
+                val message = HL7Reader.parseHL7Message(requestBody)
                 val acceptAcknowledgementType = HL7Reader.getAcceptAcknowledgmentType(message)
                 val ackResponseRequired = acceptAcknowledgmentTypeRespondValues.contains(acceptAcknowledgementType)
                 if (ackResponseRequired) {

prime-router/src/main/kotlin/cli/ProcessFhirCommands.kt

@@ -223,8 +223,8 @@ class ProcessFhirCommands : CliktCommand(
                 (isCli && outputFormat == MimeFormat.HL7.toString()) ||
                     (
                         receiver != null &&
-                        (receiver.format == MimeFormat.HL7 || receiver.format == MimeFormat.HL7_BATCH)
-                    )
+                            (receiver.format == MimeFormat.HL7 || receiver.format == MimeFormat.HL7_BATCH)
+                        )
                 ) -> {
                 val (bundle2, inputMessage) = convertHl7ToFhir(contents, receiver)
 
@@ -297,7 +297,7 @@ class ProcessFhirCommands : CliktCommand(
         }
     }
 
-        private fun evaluateReceiverFilters(receiver: Receiver?, messageOrBundle: MessageOrBundle, isCli: Boolean) {
+    private fun evaluateReceiverFilters(receiver: Receiver?, messageOrBundle: MessageOrBundle, isCli: Boolean) {
         if (receiver != null && messageOrBundle.bundle != null) {
             val reportStreamFilters = mutableListOf<Pair<String, ReportStreamFilter>>()
             reportStreamFilters.add(Pair("Jurisdictional Filter", receiver.jurisdictionalFilter))
@@ -516,16 +516,15 @@ class ProcessFhirCommands : CliktCommand(
      * look like.
      * @return a FHIR bundle and the parsed HL7 input that represents the data in the one HL7 message
      */
+    @Suppress("DEPRECATION")
     private fun convertHl7ToFhir(hl7String: String, receiver: Receiver?): Pair<Bundle, Message> {
         val hasFiveEncodingChars = hl7MessageHasFiveEncodingChars(hl7String)
         // Some HL7 2.5.1 implementations have adopted the truncation character # that was added in 2.7
         // However, the library used to encode the HL7 message throws an error it there are more than 4 encoding
         // characters, so this work around exists for that scenario
         val stringToEncode = hl7String.replace("MSH|^~\\&#|", "MSH|^~\\&|")
-        val hl7message = HL7Reader.parseHL7Message(
-            stringToEncode,
-            null
-        )
+        val hl7message = HL7Reader.parseHL7Message(stringToEncode)
+
         // if a hl7 parsing failure happens, throw error and show the message
         if (hl7message.toString().lowercase().contains("failed")) {
             throw CliktError("HL7 parser failure. $hl7message")
@@ -534,12 +533,8 @@ class ProcessFhirCommands : CliktCommand(
             val msh = hl7message.get("MSH") as Segment
             Terser.set(msh, 2, 0, 1, 1, "^~\\&#")
         }
-        val hl7profile = HL7Reader.getMessageProfile(hl7message.toString())
         // search hl7 profile map and create translator with config path if found
-        var fhirMessage = when (val configPath = HL7Reader.profileDirectoryMap[hl7profile]) {
-            null -> HL7toFhirTranslator(inputSchema).translate(hl7message)
-            else -> HL7toFhirTranslator(configPath).translate(hl7message)
-        }
+        var fhirMessage = HL7toFhirTranslator(inputSchema).translate(hl7message)
 
         val stamper = ConditionStamper(LookupTableConditionMapper(Metadata.getInstance()))
         fhirMessage.getObservations().forEach { observation ->

prime-router/src/main/kotlin/cli/ProcessHl7Commands.kt

@@ -1,11 +1,11 @@
 package gov.cdc.prime.router.cli
 
+import ca.uhn.hl7v2.util.Hl7InputStreamMessageStringIterator
 import com.github.ajalt.clikt.core.CliktCommand
 import com.github.ajalt.clikt.core.CliktError
 import com.github.ajalt.clikt.parameters.options.option
 import com.github.ajalt.clikt.parameters.options.required
 import com.github.ajalt.clikt.parameters.types.file
-import gov.cdc.prime.router.ActionLogger
 import gov.cdc.prime.router.cli.helpers.HL7DiffHelper
 import gov.cdc.prime.router.fhirengine.utils.HL7Reader
 
@@ -45,9 +45,14 @@ class ProcessHl7Commands : CliktCommand(
         val comparisonFile = comparisonFile.inputStream().readBytes().toString(Charsets.UTF_8)
         if (comparisonFile.isBlank()) throw CliktError("File ${this.comparisonFile.absolutePath} is empty.")
 
-        val actionLogger = ActionLogger()
-        val starterMessages = HL7Reader(actionLogger).getMessages(starterFile)
-        val comparisonMessages = HL7Reader(actionLogger).getMessages(comparisonFile)
+        val starterMessages = Hl7InputStreamMessageStringIterator(starterFile.byteInputStream()).asSequence()
+            .map { rawItem ->
+                HL7Reader.parseHL7Message(rawItem)
+            }.toList()
+        val comparisonMessages = Hl7InputStreamMessageStringIterator(comparisonFile.byteInputStream()).asSequence()
+            .map { rawItem ->
+                HL7Reader.parseHL7Message(rawItem)
+            }.toList()
 
         starterMessages.forEachIndexed { counter, message ->
             val differences = hl7DiffHelper.diffHl7(message, comparisonMessages[counter])

prime-router/src/main/kotlin/fhirengine/engine/FHIRConverter.kt

@@ -50,11 +50,9 @@ import gov.cdc.prime.router.fhirengine.translation.hl7.FhirTransformer
 import gov.cdc.prime.router.fhirengine.translation.hl7.utils.CustomContext
 import gov.cdc.prime.router.fhirengine.translation.hl7.utils.FhirPathUtils
 import gov.cdc.prime.router.fhirengine.utils.FhirTranscoder
-import gov.cdc.prime.router.fhirengine.utils.HL7Reader
 import gov.cdc.prime.router.fhirengine.utils.HL7Reader.Companion.parseHL7Message
 import gov.cdc.prime.router.fhirengine.utils.getObservations
 import gov.cdc.prime.router.fhirengine.utils.getRSMessageType
-import gov.cdc.prime.router.fhirengine.utils.isElr
 import gov.cdc.prime.router.logging.LogMeasuredTime
 import gov.cdc.prime.router.report.ReportService
 import gov.cdc.prime.router.validation.IItemValidator
@@ -397,7 +395,7 @@ class FHIRConverter(
                                 )
                             }
 
-                        FHIREngineRunResult(
+                            FHIREngineRunResult(
                                 routeEvent,
                                 report,
                                 blobInfo.blobUrl,
@@ -427,7 +425,7 @@ class FHIRConverter(
                     report,
                     TaskAction.convert,
                     "Submitted report was either empty or could not be parsed into HL7"
-                    ) {
+                ) {
                     parentReportId(input.reportId)
                     params(
                         mapOf(
@@ -479,7 +477,7 @@ class FHIRConverter(
                                 "format" to format.name
                             )
                         ) {
-                            getBundlesFromRawHL7(rawReport, validator, input.topic.hl7ParseConfiguration)
+                            getBundlesFromRawHL7(rawReport, validator)
                         }
                     } catch (ex: ParseFailureError) {
                         actionLogger.error(
@@ -571,7 +569,6 @@ class FHIRConverter(
     private fun getBundlesFromRawHL7(
         rawReport: String,
         validator: IItemValidator,
-        hL7MessageParseAndConvertConfiguration: HL7Reader.Companion.HL7MessageParseAndConvertConfiguration?,
     ): List<IProcessedItem<Message>> {
         val itemStream =
             Hl7InputStreamMessageStringIterator(rawReport.byteInputStream()).asSequence()
@@ -580,17 +577,16 @@ class FHIRConverter(
                 }.toList()
 
         return maybeParallelize(itemStream.size, itemStream.stream(), "Generating FHIR bundles in").map { item ->
-            parseHL7Item(item, hL7MessageParseAndConvertConfiguration)
+            parseHL7Item(item)
         }.map { item ->
-            validateAndConvertHL7Item(item, validator, hL7MessageParseAndConvertConfiguration)
+            validateAndConvertHL7Item(item, validator)
         }.collect(Collectors.toList())
     }
 
     private fun parseHL7Item(
         item: ProcessedHL7Item,
-        hL7MessageParseAndConvertConfiguration: HL7Reader.Companion.HL7MessageParseAndConvertConfiguration?,
     ) = try {
-        val message = parseHL7Message(item.rawItem, hL7MessageParseAndConvertConfiguration)
+        val message = parseHL7Message(item.rawItem)
         item.updateParsed(message)
     } catch (e: HL7Exception) {
         item.updateParsed(
@@ -605,20 +601,11 @@ class FHIRConverter(
     private fun validateAndConvertHL7Item(
         item: ProcessedHL7Item,
         validator: IItemValidator,
-        hL7MessageParseAndConvertConfiguration: HL7Reader.Companion.HL7MessageParseAndConvertConfiguration?,
     ): ProcessedHL7Item = if (item.parsedItem != null) {
         val validationResult = validator.validate(item.parsedItem)
         if (validationResult.isValid()) {
             try {
-                val bundle = when (hL7MessageParseAndConvertConfiguration) {
-                    null -> HL7toFhirTranslator.getHL7ToFhirTranslatorInstance().translate(item.parsedItem)
-                    else ->
-                        HL7toFhirTranslator
-                            .getHL7ToFhirTranslatorInstance(
-                                hL7MessageParseAndConvertConfiguration.hl7toFHIRMappingLocation
-                            )
-                            .translate(item.parsedItem)
-                }
+                val bundle = HL7toFhirTranslator.getHL7ToFhirTranslatorInstance().translate(item.parsedItem)
                 item.setBundle(bundle)
             } catch (ex: Exception) {
                 item.setConversionError(
@@ -759,13 +746,13 @@ class FHIRConverter(
      * transformer in tests.
      */
     fun getTransformerFromSchema(schemaName: String): FhirTransformer? = if (schemaName.isNotBlank()) {
-            withLoggingContext(mapOf("schemaName" to schemaName)) {
-                logger.info("Apply a sender transform to the items in the report")
-            }
-            FhirTransformer(schemaName)
-        } else {
-            null
+        withLoggingContext(mapOf("schemaName" to schemaName)) {
+            logger.info("Apply a sender transform to the items in the report")
         }
+        FhirTransformer(schemaName)
+    } else {
+        null
+    }
 }
 
 /**

prime-router/src/main/kotlin/fhirengine/translation/TranslationSchemaManager.kt

@@ -1,9 +1,9 @@
 package gov.cdc.prime.router.fhirengine.translation
 
+import ca.uhn.hl7v2.util.Hl7InputStreamMessageStringIterator
 import com.azure.storage.blob.models.BlobItem
 import fhirengine.engine.CustomFhirPathFunctions
 import fhirengine.engine.CustomTranslationFunctions
-import gov.cdc.prime.router.ActionLogger
 import gov.cdc.prime.router.Hl7Configuration
 import gov.cdc.prime.router.azure.BlobAccess
 import gov.cdc.prime.router.fhirengine.config.HL7TranslationConfig
@@ -41,7 +41,6 @@ class TranslationSchemaManager : Logging {
             Regex("/$previousValidBlobName-$timestampRegex")
         private val previousPreviousValidBlobNameRegex =
             Regex("/$previousPreviousValidBlobName-$timestampRegex")
-        private val hL7Reader = HL7Reader(ActionLogger())
 
         /**
          * Container class that holds the current state for a schema type in a particular azure store.
@@ -440,7 +439,11 @@ class TranslationSchemaManager : Logging {
                             )
                         ).validate(
                             inputBundle,
-                            hL7Reader.getMessages(rawValidationInput.output)[0]
+                            HL7Reader.parseHL7Message(
+                                Hl7InputStreamMessageStringIterator(rawValidationInput.output.byteInputStream())
+                                    .asSequence()
+                                    .first()
+                            ),
                         )
                     }
                 }

prime-router/src/main/kotlin/fhirengine/utils/HL7MessageHelpers.kt

@@ -1,6 +1,8 @@
 package gov.cdc.prime.router.fhirengine.utils
 
+import ca.uhn.hl7v2.AbstractHL7Exception
 import ca.uhn.hl7v2.model.v251.datatype.DTM
+import ca.uhn.hl7v2.util.Hl7InputStreamMessageStringIterator
 import ca.uhn.hl7v2.util.Terser
 import gov.cdc.prime.router.ActionLogger
 import gov.cdc.prime.router.Hl7Configuration
@@ -24,6 +26,8 @@ object HL7MessageHelpers : Logging {
      */
     const val hl7SegmentDelimiter = "\r"
 
+    val actionLogger = ActionLogger()
+
     /**
      * Generate a HL7 Batch file from the list of [hl7RawMsgs] for the given [receiver].  The [hl7RawMsgs] are expected
      * to be real HL7 messages at this point, so we will not validate their contents here for performance reasons.
@@ -34,12 +38,17 @@ object HL7MessageHelpers : Logging {
         val useBatchHeaders = receiver.translation.useBatchHeaders
         // Grab the first message to extract some data if not set in the settings
         val firstMessage = if (hl7RawMsgs.isNotEmpty()) {
-            val messages = HL7Reader(ActionLogger()).getMessages(hl7RawMsgs[0])
-            if (messages.isEmpty()) {
+            try {
+                val message = HL7Reader.parseHL7Message(hl7RawMsgs[0])
+                Terser(message)
+            } catch (exception: Hl7InputStreamMessageStringIterator.ParseFailureError) {
+                logger.warn("Unable to extract batch header values from HL7: ${hl7RawMsgs[0].take(80)} ...")
+                HL7Reader.logHL7ParseFailure(exception, actionLogger)
+                null
+            } catch (exception: AbstractHL7Exception) {
                 logger.warn("Unable to extract batch header values from HL7: ${hl7RawMsgs[0].take(80)} ...")
+                HL7Reader.recordError(exception, actionLogger)
                 null
-            } else {
-                Terser(messages[0])
             }
         } else {
             null
@@ -94,4 +103,8 @@ object HL7MessageHelpers : Logging {
 
         return builder.toString()
     }
+
+    fun messageCount(rawHl7: String): Int {
+        return Hl7InputStreamMessageStringIterator(rawHl7.byteInputStream()).asSequence().count()
+    }
 }