Threaded client message reader

src/protocol/PasLS.JSONSignaledQueue.pas

@@ -0,0 +1,120 @@
+unit PasLS.JSONSignaledQueue;
+
+{$mode objfpc}{$H+}
+
+interface
+
+uses
+  SysUtils, fpjson, fgl, syncobjs, Classes;
+
+type
+  TJSONSignaledQueue = class
+  private
+    FList: specialize TFPGObjectList<TJSONData>;
+    FLock: TCriticalSection;
+    FReadyEvent: TEvent; // Event used to signal new data inQueue
+  public
+    constructor Create;
+    destructor Destroy; override;
+
+    procedure Enqueue(AItem: TJSONData);
+    { DequeueWaiting blocks the calling thread until an item is available 
+      or the timeout expires. Returns True if an item was successfully retrieved. }
+    function DequeueWaiting(out AItem: TJSONData; TimeoutMS: Cardinal = INFINITE): Boolean;
+    procedure SignalShutdown;
+  end;
+
+implementation
+
+constructor TJSONSignaledQueue.Create;
+begin
+  inherited Create;
+  // FList owns request JSON Data
+  FList := specialize TFPGObjectList<TJSONData>.Create(True);
+  FLock := TCriticalSection.Create;
+  // ManualReset = False (auto-resets when a thread wakes up)
+  // InitialState = False (starts unsignaled)
+  FReadyEvent := TEvent.Create(nil, False, False, '');
+end;
+
+destructor TJSONSignaledQueue.Destroy;
+begin
+  FList.Free;
+  FReadyEvent.Free;
+  FLock.Free;
+  inherited Destroy;
+end;
+
+procedure TJSONSignaledQueue.Enqueue(AItem: TJSONData);
+var
+  I: Integer;
+  JSONObj: TJSONObject;
+begin
+  FLock.Acquire;
+  try
+    if AItem is TJSONObject then
+      begin
+        JSONObj := (AItem as TJSONObject);
+        if JSONObj.Get('method', '') = 'textDocument/didChange' then
+          begin
+            for I := 0 to FList.Count - 1 do
+              begin
+                if Flist[I] is TJsonObject then
+                  begin
+                    if (Flist[I] as TJSONObject).Get('method', '') = 'textDocument/didChange' then
+                      begin
+                        FList.Remove(FList[I]);
+                        break
+                      end;
+                  end;
+              end;
+          end;
+      end;    
+      FList.Add(AItem);
+  finally
+    FLock.Release;
+  end;
+  // Signal DequeueWaiting to resume waiting
+  FReadyEvent.SetEvent; 
+end;
+
+function TJSONSignaledQueue.DequeueWaiting(out AItem: TJSONData; TimeoutMS: Cardinal): Boolean;
+var
+  WaitResult: TWaitResult;
+begin
+  AItem := nil;
+  Result := False;
+
+  while True do
+  begin
+    FLock.Acquire;
+    try
+      if FList.Count > 0 then
+      begin
+        AItem := FList[0];
+        // Effectively transfers ownership to AItem
+        FList.Extract(AItem);
+        Result := True;
+      end;
+    finally
+      FLock.Release;
+    end;
+    if Result then
+      Exit; // Successfully retrieved an item
+
+    // Queue was empty; sleep efficiently until an item is pushed or timeout occurs
+    WaitResult := FReadyEvent.WaitFor(TimeoutMS);
+
+    // If it timed out or the event was explicitly reset/abandoned, exit gracefully
+    if WaitResult <> wrSignaled then
+      Exit;
+  end;
+end;
+
+procedure TJSONSignaledQueue.SignalShutdown;
+begin
+  // Broadcast to all waiting threads so they wake up and check terminate status
+  FReadyEvent.SetEvent; 
+end;
+
+end.

src/protocol/PasLS.TextLoop.pas

@@ -24,7 +24,7 @@
 interface
 
 uses
-  Classes, SysUtils, LSP.Base, LSP.Messages, fpjson;
+  Classes, SysUtils, LSP.Base, LSP.Messages, fpjson, PasLS.JSONSignaledQueue;
 
 Type
 
@@ -44,7 +44,16 @@   TLSPTextTransport = class(TMessageTransport)
     Procedure EmitMessage(aMessage: TJSONStringType);
   end;
 
-
+  TRequestReaderThread = class(TThread)
+  private
+    FQueue: TJSONSignaledQueue;
+    FFile: ^Text;
+    FContext: ^TLSPContext;
+  protected
+    procedure Execute; override;
+  public
+    constructor Create(AQueue: TJSONSignaledQueue; var aFile : text; var aContext : TLSPContext);
+  end;
 
 Procedure SetupTextLoop(var aInput,aOutput,aError : Text);
 Procedure RunMessageLoop(var aInput,aOutput,aError : Text; aContext : TLSPContext);
@@ -89,40 +98,56 @@ procedure DebugSendMessage(var aFile : Text; aContext : TLSPContext; const aMeth
 end;
 
 
+constructor TRequestReaderThread.Create(AQueue: TJSONSignaledQueue; var aFile : text; var aContext : TLSPContext);
+begin
+  inherited Create(False); // Start immediately
+  FQueue := AQueue;
+  FreeOnTerminate := False;
 
-Function ReadRequest(var aFile : text; aContext : TLSPContext) : TJSONData;
+  FFile := @aFile;
+  FContext := @aContext;
+end;
 
+procedure TRequestReaderThread.Execute;
 Var
   Header,Name,Value: String;
   Content : TJSONStringType;
   I,ContentLength : Integer;
   P : PJSONCharType;
+  ContentData: TJSONData;
 
 begin
-  Result:=Nil;
-  aContext.Log('Reading request');
-  ReadLn(aFile,Header);
-  while Header <> '' do
+  while not Terminated and not EOF(FFile^) do
     begin
-      aContext.Log('Read header: %s',[Header]);
-      I := Pos(':', Header);
-      Name := Copy(Header, 1, I - 1);
-      Delete(Header, 1, i);
-      Value := Trim(Header);
-      if Name = 'Content-Length' then
-        ContentLength := StrToIntDef(Value,0);
-      ReadLn(aFile,Header);
-    end;
-  Content:='';
-  SetLength(Content,ContentLength);
-  P:=PJSONCharType(Content);
-  for I:=1 to ContentLength do
-    begin
-    Read(aFile,P^);
-    inc(P);
+      FContext^.Log('Reading request');
+      ReadLn(FFile^,Header);
+
+      while Header <> '' do
+        begin
+          FContext^.Log('Read header: %s',[Header]);
+          I := Pos(':', Header);
+          Name := Copy(Header, 1, I - 1);
+          Delete(Header, 1, i);
+          Value := Trim(Header);
+          if Name = 'Content-Length' then
+            ContentLength := StrToIntDef(Value,0);
+          ReadLn(FFile^,Header);
+        end;
+      Content:='';
+      SetLength(Content,ContentLength);
+      P:=PJSONCharType(Content);
+      for I:=1 to ContentLength do
+        begin
+        Read(FFile^,P^);
+        inc(P);
+        end;
+      if Content<>'' then
+        begin
+          ContentData := GetJSON(Content, True);
+          FQueue.Enqueue(ContentData);
+        end;
+      FContext^.Log('Waiting request');
     end;
-  if Content<>'' then
-    Result:=GetJSON(Content, True);
 end;
 
 Procedure SendResponse(aTransport : TMessageTransport; aContext : TLSPContext; aResponse : TJSONData; aFreeResponse : Boolean = True);
@@ -150,6 +175,8 @@ procedure DebugSendMessage(var aFile : Text; aContext : TLSPContext; const aMeth
 Procedure RunMessageLoop(var aInput,aOutput,aError : Text; aContext : TLSPContext);
 
 var
+  RequestQueue: TJSONSignaledQueue;
+  RequestReader: TRequestReaderThread;
   Request, Response: TJSONData;
   VerboseDebugging: boolean = false;
   IO : TLSPTextTransport;
@@ -158,38 +185,48 @@ procedure DebugSendMessage(var aFile : Text; aContext : TLSPContext; const aMeth
   IO:=Nil;
   Request:=Nil;
   try
+    RequestQueue := TJSONSignaledQueue.Create;
+    RequestReader := TRequestReaderThread.Create(RequestQueue, aInput, aContext);
+
     if aContext.Transport is TLSPTextTransport then
       IO:=aContext.Transport as TLSPTextTransport
     else
       IO:=TLSPTextTransport.Create(@aOutput,@aError);
-    while not EOF(aInput) do
+    while not RequestReader.Finished do
       begin
-      Request:=ReadRequest(aInput,aContext);
-      // log request payload
-      if VerboseDebugging then
-        begin
-          Writeln(aError, Request.FormatJSON);
-          Flush(aError);
-        end;
-      Response := aContext.Execute(Request);
-      if Assigned(Response) then
-        begin
-        // log response payload
-        if VerboseDebugging then
+        if RequestQueue.DequeueWaiting(Request, $FFFFFFFF) then
           begin
-          writeln(aError, Response.asJSON);
-          Flush(aError);
+            // log request payload
+            if VerboseDebugging then
+              begin
+                Writeln(aError, Request.FormatJSON);
+                Flush(aError);
+              end;
+            Response := aContext.Execute(Request);
+            if Assigned(Response) then
+              begin
+              // log response payload
+              if VerboseDebugging then
+                begin
+                writeln(aError, Response.asJSON);
+                Flush(aError);
+                end;
+              SendResponse(IO, aContext, Response,True);
+              end
+            else
+              aContext.Log('No response to request');
+            FreeAndNil(Request);
           end;
-        SendResponse(IO, aContext, Response,True);
-        end
-      else
-        aContext.Log('No response to request');
-      FreeAndNil(Request);
       end;
   finally
     if IO<>aContext.Transport then
       IO.Free;
     Request.Free;
+    RequestReader.Terminate;
+    RequestQueue.SignalShutdown;
+    RequestReader.WaitFor;
+    RequestReader.Free;
+    RequestQueue.Free;
   end;
 end;
 

src/standard/pasls.lpr

@@ -23,6 +23,9 @@
 {$mode objfpc}{$H+}
 
 uses
+  {$IFDEF UNIX}
+  cthreads, // Required for multi-threading on Linux/macOS
+  {$ENDIF}
   {$IFDEF WINDOWS}
   Windows,
   {$ENDIF}