Postgresql adapter

host/MR.AspNetCore.Jobs.EFHost/MR.AspNetCore.Jobs.EFHost.csproj

@@ -7,6 +7,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\..\src\MR.AspNetCore.Jobs.SqlServer\MR.AspNetCore.Jobs.SqlServer.csproj" />
+    <ProjectReference Include="..\..\src\MR.AspNetCore.Jobs.PostgreSQL\MR.AspNetCore.Jobs.PostgreSQL.csproj" />
   </ItemGroup>
 
 </Project>

src/MR.AspNetCore.Jobs.EFCore/EFCoreStorageConnection.cs

@@ -194,6 +194,11 @@ private async Task<IFetchedJob> FetchNextDelayedJobUsingTimeoutAsync(string sql)
 				return null;
 			}
 
+			return CreateSqlTimeoutFetchedJob(fetchedJob);
+		}
+
+		protected virtual IFetchedJob CreateSqlTimeoutFetchedJob(FetchedJob fetchedJob)
+		{
 			return new SqlTimeoutFetchedJob(
 				Services,
 				fetchedJob.Id,

src/MR.AspNetCore.Jobs.EFCore/Server/SqlTimeoutFetchedJob.cs

@@ -44,7 +44,7 @@ public Task RemoveFromQueueAsync()
 					var connection = storageConnection.GetDbConnection();
 
 					connection.Execute(
-						$"DELETE FROM {storageConnection.BaseOptions.Schema}.JobQueue WHERE Id = @id",
+						CreateRemoveFromQueueQuery(storageConnection),
 						new { id = Id });
 
 					_removedFromQueue = true;
@@ -54,6 +54,11 @@ public Task RemoveFromQueueAsync()
 			return Task.CompletedTask;
 		}
 
+		protected virtual string CreateRemoveFromQueueQuery(EFCoreStorageConnection storageConnection)
+		{
+			return $"DELETE FROM {storageConnection.BaseOptions.Schema}.JobQueue WHERE Id = @id";
+		}
+
 		public Task RequeueAsync()
 		{
 			lock (_lock)
@@ -64,7 +69,7 @@ public Task RequeueAsync()
 					var connection = storageConnection.GetDbConnection();
 
 					connection.Execute(
-						$"UPDATE {storageConnection.BaseOptions.Schema}.JobQueue SET FetchedAt = null WHERE Id = @id",
+						CreateRequeueQuery(storageConnection),
 						new { id = Id });
 
 					_requeued = true;
@@ -74,6 +79,11 @@ public Task RequeueAsync()
 			return Task.CompletedTask;
 		}
 
+		protected virtual string CreateRequeueQuery(EFCoreStorageConnection storageConnection)
+		{
+			return $"UPDATE {storageConnection.BaseOptions.Schema}.JobQueue SET FetchedAt = NULL WHERE Id = @id";
+		}
+
 		public void Dispose()
 		{
 			if (_disposed) return;

src/MR.AspNetCore.Jobs.PostgreSQL/JobsDbContextFactory.cs

@@ -0,0 +1,33 @@
+using System;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Design;
+using Microsoft.Extensions.DependencyInjection;
+using MR.AspNetCore.Jobs.Models;
+
+namespace MR.AspNetCore.Jobs
+{
+	public class JobsDbContextFactory : IDesignTimeDbContextFactory<JobsDbContext>
+	{
+		internal static string DevConnectionString =
+			Environment.GetEnvironmentVariable("MR_ASPNETCORE_JOBS_POSTGRESQL_CS_DEV") ??
+			@"Server=127.0.0.1;Port=5432;Database=MR.AspNetCore.Jobs.Dev;User Id=postgres;Password=password;";
+
+		public JobsDbContext CreateDbContext(string[] args)
+		{
+			var services = new ServiceCollection();
+
+			services.AddSingleton(new PostgreSQLOptions());
+			services.AddDbContext<JobsDbContext>(opts =>
+			{
+				opts.UseNpgsql(DevConnectionString, sqlOpts =>
+				{
+					sqlOpts.MigrationsHistoryTable(
+						EFCoreOptions.DefaultMigrationsHistoryTableName,
+						EFCoreOptions.DefaultSchema);
+				});
+			});
+
+			return services.BuildServiceProvider().GetRequiredService<JobsDbContext>();
+		}
+	}
+}

src/MR.AspNetCore.Jobs.PostgreSQL/LoggerExtensions.cs

@@ -0,0 +1,6 @@
+namespace MR.AspNetCore.Jobs
+{
+	internal static class LoggerExtensions
+	{
+	}
+}

src/MR.AspNetCore.Jobs.PostgreSQL/MR.AspNetCore.Jobs.PostgreSQL.csproj

@@ -0,0 +1,29 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <Import Project="../../build/common.props" />
+
+  <PropertyGroup Label="Build">
+    <TargetFrameworks>netstandard2.0</TargetFrameworks>
+    <RootNamespace>MR.AspNetCore.Jobs</RootNamespace>
+    <AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects>
+    <GenerateBindingRedirectsOutputType>true</GenerateBindingRedirectsOutputType>
+  </PropertyGroup>
+
+  <PropertyGroup Label="Package">
+    <Description>A PostgreSQL adapter for MR.AspNetCore.Jobs.</Description>
+    <PackageTags>aspnetcore;background;jobs;sql;postgresql;netstandard</PackageTags>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\MR.AspNetCore.Jobs.EFCore\MR.AspNetCore.Jobs.EFCore.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="2.0.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <DotNetCliToolReference Include="Microsoft.EntityFrameworkCore.Tools.DotNet" Version="2.0.1" />
+  </ItemGroup>
+
+</Project>

src/MR.AspNetCore.Jobs.PostgreSQL/Migrations/20171125160624_InitialCreate.cs

@@ -0,0 +1,112 @@
+using Microsoft.EntityFrameworkCore.Metadata;
+using Microsoft.EntityFrameworkCore.Migrations;
+using System;
+using System.Collections.Generic;
+
+namespace MR.AspNetCore.Jobs.Migrations
+{
+    public partial class InitialCreate : Migration
+    {
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.EnsureSchema(
+                name: "Jobs");
+
+            migrationBuilder.CreateTable(
+                name: "CronJobs",
+                schema: "Jobs",
+                columns: table => new
+                {
+                    Id = table.Column<string>(nullable: false),
+                    Cron = table.Column<string>(nullable: true),
+                    LastRun = table.Column<DateTime>(nullable: false),
+                    Name = table.Column<string>(nullable: false),
+                    TypeName = table.Column<string>(nullable: true)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_CronJobs", x => x.Id);
+                });
+
+            migrationBuilder.CreateTable(
+                name: "Jobs",
+                schema: "Jobs",
+                columns: table => new
+                {
+                    Id = table.Column<int>(nullable: false)
+                        .Annotation("Npgsql:ValueGenerationStrategy", NpgsqlValueGenerationStrategy.SerialColumn),
+                    Added = table.Column<DateTime>(nullable: false),
+                    Data = table.Column<string>(nullable: true),
+                    Due = table.Column<DateTime>(nullable: true),
+                    ExpiresAt = table.Column<DateTime>(nullable: true),
+                    Retries = table.Column<int>(nullable: false),
+                    StateName = table.Column<string>(nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_Jobs", x => x.Id);
+                });
+
+            migrationBuilder.CreateTable(
+                name: "JobQueue",
+                schema: "Jobs",
+                columns: table => new
+                {
+                    Id = table.Column<int>(nullable: false)
+                        .Annotation("Npgsql:ValueGenerationStrategy", NpgsqlValueGenerationStrategy.SerialColumn),
+                    JobId = table.Column<int>(nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_JobQueue", x => x.Id);
+                    table.ForeignKey(
+                        name: "FK_JobQueue_Jobs_JobId",
+                        column: x => x.JobId,
+                        principalSchema: "Jobs",
+                        principalTable: "Jobs",
+                        principalColumn: "Id",
+                        onDelete: ReferentialAction.Cascade);
+                });
+
+            migrationBuilder.CreateIndex(
+                name: "IX_CronJobs_Name",
+                schema: "Jobs",
+                table: "CronJobs",
+                column: "Name",
+                unique: true);
+
+            migrationBuilder.CreateIndex(
+                name: "IX_JobQueue_JobId",
+                schema: "Jobs",
+                table: "JobQueue",
+                column: "JobId");
+
+            migrationBuilder.CreateIndex(
+                name: "IX_Jobs_StateName",
+                schema: "Jobs",
+                table: "Jobs",
+                column: "StateName");
+
+            migrationBuilder.CreateIndex(
+                name: "IX_Jobs_Due_StateName",
+                schema: "Jobs",
+                table: "Jobs",
+                columns: new[] { "Due", "StateName" });
+        }
+
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "CronJobs",
+                schema: "Jobs");
+
+            migrationBuilder.DropTable(
+                name: "JobQueue",
+                schema: "Jobs");
+
+            migrationBuilder.DropTable(
+                name: "Jobs",
+                schema: "Jobs");
+        }
+    }
+}