Platform/angela/14116/read hl7

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

@@ -1,5 +1,6 @@
 package gov.cdc.prime.router
 
+import ca.uhn.hl7v2.util.Hl7InputStreamMessageStringIterator
 import gov.cdc.prime.reportstream.shared.BlobUtils
 import gov.cdc.prime.reportstream.shared.QueueMessage
 import gov.cdc.prime.router.azure.ActionHistory
@@ -268,14 +269,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 = Hl7InputStreamMessageStringIterator(content.byteInputStream()).asSequence().count()
+                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,12 +291,14 @@ 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."))
                 }
+                // check for valid message type
+                Hl7InputStreamMessageStringIterator(content.byteInputStream()).asSequence()
+                    .forEachIndexed { index, rawItem ->
+                        MessageType.validateMessageType(HL7Reader.parseHL7Message(rawItem, null), actionLogs, index + 1)
+                    }
             }
 
             MimeFormat.FHIR -> {

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

@@ -1,11 +1,11 @@
 package gov.cdc.prime.router.azure.service
 
 import ca.uhn.hl7v2.model.Message
+import ca.uhn.hl7v2.util.Hl7InputStreamMessageStringIterator
 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
@@ -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 = Hl7InputStreamMessageStringIterator(requestBody.byteInputStream()).asSequence().count()
+            val isBatch = HL7Reader.isBatch(requestBody, messageCount)
 
-            if (!isBatch && messages.size == 1) {
-                val message = messages.first()
+            if (!isBatch && messageCount == 1) {
+                val message = HL7Reader.parseHL7Message(requestBody, null)
                 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,6 +516,7 @@ 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

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, null)
+            }.toList()
+        val comparisonMessages = Hl7InputStreamMessageStringIterator(comparisonFile.byteInputStream()).asSequence()
+            .map { rawItem ->
+                HL7Reader.parseHL7Message(rawItem, null)
+            }.toList()
 
         starterMessages.forEachIndexed { counter, message ->
             val differences = hl7DiffHelper.diffHl7(message, comparisonMessages[counter])

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

@@ -397,7 +397,7 @@ class FHIRConverter(
                                 )
                             }
 
-                        FHIREngineRunResult(
+                            FHIREngineRunResult(
                                 routeEvent,
                                 report,
                                 blobInfo.blobUrl,
@@ -427,7 +427,7 @@ class FHIRConverter(
                     report,
                     TaskAction.convert,
                     "Submitted report was either empty or could not be parsed into HL7"
-                    ) {
+                ) {
                     parentReportId(input.reportId)
                     params(
                         mapOf(
@@ -759,13 +759,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,12 @@ class TranslationSchemaManager : Logging {
                             )
                         ).validate(
                             inputBundle,
-                            hL7Reader.getMessages(rawValidationInput.output)[0]
+                            HL7Reader.parseHL7Message(
+                                Hl7InputStreamMessageStringIterator(rawValidationInput.output.byteInputStream())
+                                    .asSequence()
+                                    .first(),
+                                null
+                            ),
                         )
                     }
                 }

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], null)
+                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