chore: introduce flix.allJobsFinished

client/src/engine/jobs.ts

@@ -52,6 +52,7 @@ export enum Request {
   internalMessage = 'ext/message', // Internal Extension Request
   internalError = 'ext/error', // Internal Extension Request
   internalFinishedJob = 'ext/finished', // Internal Extension Request
+  internalFinishedAllJobs = 'ext/finishedAll', // Internal Extension Request
   internalDiagnostics = 'ext/diagnostics', // Internal Extension Request
   internalReplaceConfiguration = 'ext/replaceConfiguration', // Internal Extension Request
 }

client/src/extension.ts

@@ -149,6 +149,12 @@ export async function activate(context: vscode.ExtensionContext, launchOptions:
   registerCommand('flix.showAst', handlers.showAst(client))
   registerCommand('flix.startRepl', handlers.startRepl(context, launchOptions))
 
+  // Register commands for testing
+
+  // Returns a promise resolving when all jobs are completely finished and the server is idle.
+  // While most other commands can be awaited directly, this is useful for stuff like file creation, which indirectely triggers an asynchronous job.
+  registerCommand('flix.allJobsFinished', handlers.allJobsFinished(client, eventEmitter))
+
   // watch for changes on the file system (delete, create, rename .flix files)
   flixWatcher = vscode.workspace.createFileSystemWatcher(FLIX_GLOB_PATTERN)
   flixWatcher.onDidDelete((vsCodeUri: vscode.Uri) => {
@@ -246,19 +252,23 @@ async function startSession(
   })
 
   // Handle when server has answered back after getting the notification above
-  client.onNotification(jobs.Request.internalReady, function handler() {
+  client.onNotification(jobs.Request.internalReady, () => {
     // waits for server to answer back after having started successfully
     eventEmitter.emit(jobs.Request.internalReady)
 
     // start the Flix runner (but only after the Flix LSP instance has started.)
     handlers.initSharedRepl(context, launchOptions)
   })
 
-  client.onNotification(jobs.Request.internalFinishedJob, function handler() {
+  client.onNotification(jobs.Request.internalFinishedJob, () => {
     // only one job runs at once, so currently not trying to distinguish
     eventEmitter.emit(jobs.Request.internalFinishedJob)
   })
 
+  client.onNotification(jobs.Request.internalFinishedAllJobs, () =>
+    eventEmitter.emit(jobs.Request.internalFinishedAllJobs),
+  )
+
   client.onNotification(jobs.Request.internalDiagnostics, handlePrintDiagnostics)
 
   client.onNotification(jobs.Request.internalRestart, makeHandleRestartClient(context))

client/src/handlers/handlers.ts

@@ -1,5 +1,6 @@
 import * as vscode from 'vscode'
 import { LanguageClient } from 'vscode-languageclient/node'
+import { EventEmitter } from 'events'
 
 import * as jobs from '../engine/jobs'
 import ensureFlixExists from './../util/ensureFlixExists'
@@ -376,3 +377,11 @@ export function startRepl(context: vscode.ExtensionContext, launchOptions: Launc
     flixTerminal?.show()
   }
 }
+
+export function allJobsFinished(client: LanguageClient, eventEmitter: EventEmitter) {
+  return () =>
+    new Promise(resolve => {
+      client.sendNotification(jobs.Request.internalFinishedAllJobs)
+      eventEmitter.once(jobs.Request.internalFinishedAllJobs, resolve)
+    })
+}

server/src/engine/flix.ts

@@ -263,3 +263,7 @@ export function enqueueJobWithFlattenedParams(request: jobs.Request, params?: an
   }
   return queue.enqueue(job)
 }
+
+export function unfinishedJobs() {
+  return queue.unfinishedJobs()
+}

server/src/engine/jobs.ts

@@ -55,6 +55,7 @@ export enum Request {
   internalMessage = 'ext/message', // Internal Extension Request
   internalError = 'ext/error', // Internal Extension Request
   internalFinishedJob = 'ext/finished', // Internal Extension Request
+  internalFinishedAllJobs = 'ext/finishedAll', // Internal Extension Request
   internalDiagnostics = 'ext/diagnostics', // Internal Extension Request
   internalReplaceConfiguration = 'ext/replaceConfiguration', // Internal Extension Request
 }

server/src/engine/queue.ts

@@ -192,3 +192,10 @@ export async function terminateQueue() {
   })
   emptyQueue()
 }
+
+/**
+ * The number of jobs which have been added to the queue, but have yet to be processed.
+ */
+export function unfinishedJobs() {
+  return priorityQueue.length + taskQueue.length + socket.unprocessedRequests()
+}

server/src/engine/socket.ts

@@ -209,4 +209,13 @@ function handleResponse(flixResponse: FlixResponse, job: jobs.EnqueuedJob) {
   clearTimer(flixResponse.id)
   // ask queue to process next item
   setTimeout(queue.processQueue, 0)
+
+  eventEmitter.emit('any')
+}
+
+/**
+ * The number of sent requests which have not yet received a response.
+ */
+export function unprocessedRequests() {
+  return Object.keys(sentMessagesMap).length
 }

server/src/handlers/handlers.ts

@@ -128,6 +128,23 @@ export const handleShowAst = enqueueUnlessHasErrors(
   hasErrorsHandlerForCommands,
 )
 
+/**
+ * Request a response to be sent when all jobs are finished.
+ */
+export function handleFinishedAllJobs() {
+  if (engine.unfinishedJobs() === 0) {
+    // If already idle, send notification immediately
+    sendNotification(jobs.Request.internalFinishedAllJobs)
+  } else {
+    socket.eventEmitter.on('any', function handler() {
+      if (engine.unfinishedJobs() === 0) {
+        sendNotification(jobs.Request.internalFinishedAllJobs)
+        socket.eventEmitter.removeListener('any', handler)
+      }
+    })
+  }
+}
+
 function makeShowAstJob(params: any) {
   return {
     request: jobs.Request.lspShowAst,

server/src/server.ts

@@ -43,6 +43,8 @@ connection.onNotification(jobs.Request.apiRemJar, handlers.handleRemJar)
 // Show ast
 connection.onNotification(jobs.Request.lspShowAst, handlers.handleShowAst)
 
+connection.onNotification(jobs.Request.internalFinishedAllJobs, handlers.handleFinishedAllJobs)
+
 // Cleanup after exit
 connection.onExit(handlers.handleExit)
 

test/src/util.ts

@@ -54,20 +54,34 @@ export async function sleep(ms: number) {
   return new Promise(resolve => setTimeout(resolve, ms))
 }
 
+/**
+ * Waits for the processing of a newly added or deleted file to finish.
+ */
+async function processFileChange() {
+  // Wait for the file system watcher to pick up the change
+  await sleep(1000)
+
+  // Wait for the compiler to process the change
+  await vscode.commands.executeCommand('flix.allJobsFinished')
+
+  // Wait for the diagnostics to be updated
+  await sleep(1000)
+}
+
 /**
  * Add a file with the given `uri` and `content`, and wait for the compiler to process this.
  */
 export async function addFile(uri: vscode.Uri, content: string) {
   await vscode.workspace.fs.writeFile(uri, Buffer.from(content))
-  await sleep(6000)
+  await processFileChange()
 }
 
 /**
  * Copy the file from `from` to `to`, and wait for the compiler to process this.
  */
 export async function copyFile(from: vscode.Uri, to: vscode.Uri) {
   await vscode.workspace.fs.copy(from, to)
-  await sleep(6000)
+  await processFileChange()
 }
 
 /**
@@ -77,5 +91,5 @@ export async function copyFile(from: vscode.Uri, to: vscode.Uri) {
  */
 export async function deleteFile(uri: vscode.Uri) {
   await vscode.workspace.fs.delete(uri)
-  await sleep(2000)
+  await processFileChange()
 }