Unify ansiLogObserver/agentLogObserver into single logObserver field

Only one observer is ever active. Use instanceof for ANSI-specific
features (appendSticky, started/stopped checks).

Co-authored-by: Ben Sherman <bentshermann@gmail.com>
Signed-off-by: Edmund Miller <edmund.miller@seqera.io>

Author: edmundmiller

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -70,7 +70,7 @@ import nextflow.script.WorkflowMetadata
 import nextflow.script.dsl.ProcessConfigBuilder
 import nextflow.spack.SpackConfig
 import nextflow.trace.AgentLogObserver
-import nextflow.trace.AnsiLogObserver
+import nextflow.trace.LogObserver
 import nextflow.trace.TraceObserver
 import nextflow.trace.TraceObserverFactory
 import nextflow.trace.TraceObserverFactoryV2
@@ -317,9 +317,7 @@ class Session implements ISession {
 
     boolean disableJobsCancellation
 
-    AnsiLogObserver ansiLogObserver
-
-    AgentLogObserver agentLogObserver
+    LogObserver logObserver
 
     FilePorter getFilePorter() { filePorter }
 
@@ -506,9 +504,10 @@ class Session implements ISession {
         result.add(statsObserver)
         // Add AgentLogObserver when agent mode is enabled
         if( agentLog ) {
-            this.agentLogObserver = new AgentLogObserver()
-            agentLogObserver.setStatsObserver(statsObserver)
-            result.add(agentLogObserver)
+            def observer = new AgentLogObserver()
+            observer.setStatsObserver(statsObserver)
+            this.logObserver = observer
+            result.add(observer)
         }
         for( TraceObserverFactoryV2 f : Plugins.getExtensions(TraceObserverFactoryV2) ) {
             log.debug "Observer factory (v2): ${f.class.simpleName}"
@@ -844,8 +843,7 @@ class Session implements ISession {
             shutdown0()
             notifyError(null)
             // force termination
-            ansiLogObserver?.forceTermination()
-            agentLogObserver?.forceTermination()
+            logObserver?.forceTermination()
             executorFactory?.signalExecutors()
             processesBarrier.forceTermination()
             monitorsBarrier.forceTermination()
@@ -1295,16 +1293,16 @@ class Session implements ISession {
     }
 
     void printConsole(String str, boolean newLine=false) {
-        if( ansiLogObserver )
-            ansiLogObserver.appendInfo(str)
+        if( logObserver )
+            logObserver.appendInfo(str)
         else if( newLine )
             System.out.println(str)
         else
             System.out.print(str)
     }
 
     void printConsole(Path file) {
-        ansiLogObserver ? ansiLogObserver.appendInfo(file.text) : Files.copy(file, System.out)
+        logObserver ? logObserver.appendInfo(file.text) : Files.copy(file, System.out)
     }
 
     private volatile ThreadPoolManager finalizePoolManager

modules/nextflow/src/main/groovy/nextflow/trace/DefaultObserverFactory.groovy

@@ -48,8 +48,9 @@ class DefaultObserverFactory implements TraceObserverFactoryV2 {
 
     protected void createAnsiLogObserver(Collection<TraceObserverV2> result) {
         if( session.ansiLog ) {
-            session.ansiLogObserver = new AnsiLogObserver()
-            result << session.ansiLogObserver
+            def observer = new AnsiLogObserver()
+            session.logObserver = observer
+            result << observer
         }
     }
 

modules/nextflow/src/main/groovy/nextflow/util/LoggerHelper.groovy

@@ -56,6 +56,7 @@ import groovyx.gpars.dataflow.DataflowReadChannel
 import groovyx.gpars.dataflow.DataflowWriteChannel
 import nextflow.Global
 import nextflow.Session
+import nextflow.trace.AnsiLogObserver
 import nextflow.cli.CliOptions
 import nextflow.cli.Launcher
 import nextflow.exception.AbortOperationException
@@ -714,33 +715,31 @@ class LoggerHelper {
             try {
                 final message = fmtEvent(event, session, false)
 
-                // Forward to AgentLogObserver when agent mode is enabled
-                final agentRenderer = session?.agentLogObserver
-                if( agentRenderer ) {
+                final observer = session?.logObserver
+                if( observer instanceof AnsiLogObserver ) {
+                    final renderer = (AnsiLogObserver)observer
+                    if( !renderer.started || renderer.stopped )
+                        System.out.println(message)
+                    else if( event.marker == STICKY )
+                        renderer.appendSticky(message)
+                    else if( event.level==Level.ERROR )
+                        renderer.appendError(message)
+                    else if( event.level==Level.WARN )
+                        renderer.appendWarning(message)
+                    else
+                        renderer.appendInfo(message)
+                }
+                else if( observer ) {
                     if( event.level==Level.ERROR )
-                        agentRenderer.appendError(message)
+                        observer.appendError(message)
                     else if( event.level==Level.WARN )
-                        agentRenderer.appendWarning(message)
+                        observer.appendWarning(message)
                     else
-                        agentRenderer.appendInfo(message)
-                    return
+                        observer.appendInfo(message)
                 }
-
-                final renderer = session?.ansiLogObserver
-                if( !renderer || !renderer.started || renderer.stopped )
+                else {
                     System.out.println(message)
-
-                else if( event.marker == STICKY )
-                    renderer.appendSticky(message)
-
-                else if( event.level==Level.ERROR )
-                    renderer.appendError(message)
-
-                else if( event.level==Level.WARN )
-                    renderer.appendWarning(message)
-
-                else
-                    renderer.appendInfo(message)
+                }
             }
             catch (Throwable e) {
                 e.printStackTrace()