Blocking command support (BLPOP/BRPOP & BLMOVE) (#356)

* wip - adding brpop * wip * wip * small fix * wip - adding updated broker * fixes * more fixes * Removing unnecessary code * Adding BLPOP * fixes * Adding comments * formatting * wip * wip * wip * Added command info + website docs for BRPOP / BLPOP * refactoring CollectionItemBroker * Added cancellation token for observer * Added timeout to tests * format * Limiting batch size in BasicListBlockingPopTest * Fix * Updating RespCommandsInfo.json * Added BLMOVE * format * Fixing tests + bugfix in CommandInfoUpdater * Added docs for collection broker * wip * wip * format * wip * Switching CollectionItemObserver back to class --------- Co-authored-by: Badrish Chandramouli <[email protected]>
microsoft · Jun 21, 2024 · b322f29 · b322f29
1 parent b503e3a
commit b322f29
Show file tree

Hide file tree

Showing 25 changed files with 1,771 additions and 294 deletions.
diff --git a/libs/host/GarnetServer.cs b/libs/host/GarnetServer.cs
@@ -28,6 +28,7 @@ public class GarnetServer : IDisposable
         private IDevice aofDevice;
         private TsavoriteLog appendOnlyFile;
         private SubscribeBroker<SpanByte, SpanByte, IKeySerializer<SpanByte>> broker;
+        private CollectionItemBroker itemBroker;
         private LogSettings logSettings, objLogSettings;
         private INamedDeviceFactory logFactory;
         private MemoryLogger initLogger;
@@ -234,6 +235,8 @@ private void InitializeServer()
                         revivificationSettings: objRevivSettings, logger: this.loggerFactory?.CreateLogger("TsavoriteKV  [obj]"));
                 if (objTotalMemorySize > 0)
                     objectStoreSizeTracker = new CacheSizeTracker(objectStore, objLogSettings, objTotalMemorySize, this.loggerFactory);
+
+                itemBroker = new CollectionItemBroker();
             }
 
             if (!opts.DisablePubSub)
@@ -279,7 +282,7 @@ private void InitializeServer()
             storeWrapper = new StoreWrapper(version, redisProtocolVersion, server, store, objectStore, objectStoreSizeTracker, customCommandManager, appendOnlyFile, opts, clusterFactory: clusterFactory, loggerFactory: loggerFactory);
 
             // Create session provider for Garnet
-            Provider = new GarnetProvider(storeWrapper, broker);
+            Provider = new GarnetProvider(storeWrapper, broker, itemBroker);
 
             // Create user facing API endpoints
             Metrics = new MetricsApi(Provider);

diff --git a/libs/server/AOF/AofProcessor.cs b/libs/server/AOF/AofProcessor.cs
@@ -73,7 +73,7 @@ public AofProcessor(
                 accessControlList: storeWrapper.accessControlList,
                 loggerFactory: storeWrapper.loggerFactory);
 
-            this.respServerSession = new RespServerSession(null, replayAofStoreWrapper, null);
+            this.respServerSession = new RespServerSession(null, replayAofStoreWrapper, null, null);
 
             var session = respServerSession.storageSession.basicContext.Session;
             basicContext = session.BasicContext;

diff --git a/libs/server/API/IGarnetApi.cs b/libs/server/API/IGarnetApi.cs
@@ -712,6 +712,7 @@ public interface IGarnetApi : IGarnetReadApi, IGarnetAdvancedApi
         /// <param name="count"></param>
         /// <returns></returns>
         GarnetStatus ListRightPop(ArgSlice key, int count, out ArgSlice[] elements);
+
         #endregion
 
         /// <summary>