Merge pull request #29 from peppy/redis-connector

Move redis helper functions to osu-queue-processor

osu.Server.QueueProcessor/ConnectionMultiplexerExtensions.cs

@@ -0,0 +1,76 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.Linq;
+using StackExchange.Redis;
+
+namespace osu.Server.QueueProcessor
+{
+    /// <summary>
+    /// Extensions to manage schema deployments on redis.
+    /// </summary>
+    public static class ConnectionMultiplexerExtensions
+    {
+        private static string allActiveSchemasKey => $"osu-queue:score-index:{Environment.GetEnvironmentVariable("ES_INDEX_PREFIX") ?? string.Empty}active-schemas";
+        private static string mainSchemaKey => $"osu-queue:score-index:{Environment.GetEnvironmentVariable("ES_INDEX_PREFIX") ?? string.Empty}schema";
+
+        /// <summary>
+        /// Add a new schema version to the active list. Note that it will not be set to the current schema (call <see cref="SetCurrentSchema"/> for that).
+        /// </summary>
+        public static bool AddActiveSchema(this ConnectionMultiplexer connection, string value)
+        {
+            return connection.GetDatabase().SetAdd(allActiveSchemasKey, value);
+        }
+
+        /// <summary>
+        /// Clears the current live schema.
+        /// </summary>
+        public static void ClearCurrentSchema(this ConnectionMultiplexer connection)
+        {
+            connection.GetDatabase().KeyDelete(mainSchemaKey);
+        }
+
+        /// <summary>
+        /// Get all active schemas (including past or future).
+        /// </summary>
+        public static string[] GetActiveSchemas(this ConnectionMultiplexer connection)
+        {
+            return connection.GetDatabase().SetMembers(allActiveSchemasKey).ToStringArray();
+        }
+
+        /// <summary>
+        /// Get the current (live) schema version.
+        /// </summary>
+        public static string GetCurrentSchema(this ConnectionMultiplexer connection)
+        {
+            return connection.GetDatabase().StringGet(mainSchemaKey).ToString() ?? string.Empty;
+        }
+
+        /// <summary>
+        /// Removes a specified schema from the active list.
+        /// </summary>
+        public static bool RemoveActiveSchema(this ConnectionMultiplexer connection, string value)
+        {
+            if (connection.GetCurrentSchema() == value)
+                throw new InvalidOperationException($"Specified schema is current. Call {nameof(ClearCurrentSchema)} first");
+
+            return connection.GetDatabase().SetRemove(allActiveSchemasKey, value);
+        }
+
+        /// <summary>
+        /// Set the current (live) schema version.
+        /// </summary>
+        /// <param name="connection"></param>
+        /// <param name="value"></param>
+        public static void SetCurrentSchema(this ConnectionMultiplexer connection, string value)
+        {
+            IDatabase database = connection.GetDatabase();
+
+            if (connection.GetActiveSchemas().All(s => s != value))
+                throw new InvalidOperationException($"Attempted to set current schema without schema being in active list. Call {nameof(AddActiveSchema)} first");
+
+            database.StringSet(mainSchemaKey, value);
+        }
+    }
+}

osu.Server.QueueProcessor/QueueProcessor.cs

@@ -46,8 +46,7 @@ public abstract class QueueProcessor<T> where T : QueueItem
 
         private readonly QueueConfiguration config;
 
-        private readonly Lazy<ConnectionMultiplexer> redis = new Lazy<ConnectionMultiplexer>(() => ConnectionMultiplexer.Connect(
-            Environment.GetEnvironmentVariable("REDIS_HOST") ?? "localhost"));
+        private readonly Lazy<ConnectionMultiplexer> redis = new Lazy<ConnectionMultiplexer>(RedisAccess.GetConnection);
 
         private IDatabase getRedisDatabase() => redis.Value.GetDatabase();
 

osu.Server.QueueProcessor/RedisAccess.cs

@@ -0,0 +1,20 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using StackExchange.Redis;
+
+namespace osu.Server.QueueProcessor
+{
+    /// <summary>
+    /// Provides access to a Redis database.
+    /// </summary>
+    public static class RedisAccess
+    {
+        /// <summary>
+        /// Retrieve a fresh Redis connection. Should be disposed after use.
+        /// </summary>
+        public static ConnectionMultiplexer GetConnection() =>
+            ConnectionMultiplexer.Connect(Environment.GetEnvironmentVariable("REDIS_HOST") ?? "localhost");
+    }
+}