ziyasal / disque.net

.Net library for the Disque distributed, in-memory message broker (i.e Redis as a job queue)

Disque.Net

A simple .Net client for the Disque in-memory distributed queue https://github.com/antirez/disque

Disque is ongoing experiment to build a distributed, in memory, message broker. Its goal is to capture the essence of the "Redis as a jobs queue" use case, which is usually implemented using blocking list operations, and move it into an ad-hoc, self-contained, scalable, and fault tolerant design, with simple to understand properties and guarantees, but still resembling Redis in terms of simplicity, performances, and implementation as a C non-blocking networked server.

Installation
NuGet - Disque.Net

Install-Package Disque.Net

List of supported methods

• Ping
• Info
• Info by Section
• AddJob
• AddJob with Parameters
• GetJob
• GetJob with Parameters
• Ackjob
• Qlen
• Qpeek
• DelJob
• Dequeue
• Enqueue
• Fastack
• Show
• Working

Ping

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string pong = q.Ping();

Info

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string info = q.Info();

Info by Section

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string info = q.Info("server");

AddJob

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string jobId = q.AddJob("myqueue", "message", 10);

AddJob with Parameters

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
JobParams jobParams = new JobParams
            {
                Replicate = 1,
                Retry = 10,
                Ttl = 20,
                Maxlen = 10,
                Delay = 10,
                Async = true
            };
string jobId = q.AddJob("myqueue", "message", 10, jobParams);

GetJob

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
List<Job> jobs = q.GetJob("myqueue");

GetJob with Parameters

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
List<Job> jobs = q.GetJob(100, 2, "myqueue");

AckJob

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string jobId = q.AddJob("myqueue", "message", 10);
long count = q.Ackjob(jobId);

Qlen

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
long qlen = q.Qlen("myqueue");

Qpeek

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
List<Job> jobs = q.Qpeek("myqueue", 2);

DelJob

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string jobId = q.AddJob("myqueue", "testJob", 10);
long count = q.DelJob(jobId);

Dequeue

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string jobId = q.AddJob("myqueue", "testJob", 10);
long count = q.Dequeue(jobId);

Enqueue

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string jobId = q.AddJob("myqueue", "testJob", 10);
long count = q.Enqueue(jobId);

Fastack

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string jobId = q.AddJob("fastack", "message", 10);
long count = q.Fastack(jobId);

Show

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
string jobId = q.AddJob("myqueue", "testJob", 10);
JobInfo info = q.Show(jobId);

Working

var q = new DisqueClient(new Uri("disque://192.168.59.103:7711"));
String jobId = q.AddJob("myqueue", "testJob", 10);
long secs = q.Working(jobId);

Bugs

If you encounter a bug, performance issue, or malfunction, please add an Issue with steps on how to reproduce the problem.

TODO

• Async methods

License

Code and documentation are available according to the MIT License (see LICENSE)

disque.net's Issues

GetJob with multiple queues does not work

q.GetJob(10, 10, "test_noexist", "test");

The test_noexist queue does not exist and therefore the expected output would be the next job from the test queue, but instead it just returns an empty list.

using telnet the following returns 10 jobs immediately
GETJOB TIMEOUT 10 COUNT 10 FROM unanalyzed_priority unanalyzed

Cannot access a closed Stream

how to fix this problem?

System.ObjectDisposedException: Cannot access a closed Stream.
at System.IO.__Error.StreamIsClosed()
at System.IO.BufferedStream.Write(Byte[] array, Int32 offset, Int32 count)
at CSRedis.Internal.RedisWriter.Write(RedisCommand command, Stream stream)
at CSRedis.Internal.RedisConnector.Call[T](RedisCommand1 command) at CSRedis.RedisClient.Write[T](RedisCommand1 command)
at Disque.Net.DisqueClient.AddJob(String queueName, String job, Int32 mstimeout)
at SGT.HOTW.DataInfo.AppLibs.DisqueUtility.AddJobToFds(Object data, String MemberFDS_Server)

using Common.Logging;
using Disque.Net;
using Newtonsoft.Json;
using StackExchange.Redis;
using System;

namespace TEST
{
	public class DisqueUnity
	{
		private DisqueClient _disqueClient = new DisqueClient(new Uri($"disque://127.0.0.1:7711"));


		public static DisqueUtility Instance { get; } = new DisqueUtility();

		
		public void AddJobToFds(dynamic data, string Member)
		{
			try
			{
				var jsondata = JsonConvert.SerializeObject(data);
				this._disqueClient.AddJob($"Group1", jsondata, 30000);  //issue point
			}
			catch (Exception ex)
			{
				this.log.Error($"Exception:{ex}");
			}
			
		}
	}
}

Crashes when acknowledging/dequeuing more than 1 job

Was trying to empty my queue as I was just testing some stuff.

var jobs = disque.GetJob(10, 1000, "test");
while (jobs.Count > 0)
{
	disque.Ackjob(jobs.Select(x => x.Id).ToArray());
	jobs = disque.GetJob(10, 1000, "test");
}

Throws:

An unhandled exception of type 'CSRedis.RedisException' occurred in csredis.dll
Additional information: BADID Invalid Job ID format.

Also crashes when I substituted 1000 for 2.

Support for .NET 4

I was downloaded the source code. It works fine with .NET 4. Could you please provide the package for .NET 4?

Thank you.

Peeking returns an invalid Job Id

When using Peek, the Id of the jobs returned have been set to the same value as the QueueName property.