I must implement thread-safe queue with buffer. According to the advice from the previous post: Thread-safe queue mechanism I resigned from Concurent Collections and used lock. Is it thread safe now? What can I improve?
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Timers;
class Program
{
public class Document : IItem
{
public Guid Id { get; set; }
public Guid MessageId { get; set; }
}
static void Main()
{
var queueProvider = new Provider();
var docs = new List<IItem>
{
new Document { Id = Guid.NewGuid()},
new Document {Id = Guid.NewGuid()},
new Document {Id = Guid.NewGuid()},
new Document {Id = Guid.NewGuid()},
new Document {Id = Guid.NewGuid()},
};
object locker = new object();
try
{
var tasks = new List<Task>();
var task1 = Task.Factory.StartNew(() =>
{
var timer1 = new Timer(1000) { Interval = 5000 };
timer1.Elapsed += (object sender, ElapsedEventArgs e) =>
{
foreach (var doc in docs)
{
if (queueProvider.TryEnqueue(doc))
{
Console.WriteLine("Enqueue: " + doc.Id + "taskId: 1");
Console.WriteLine("Count: " + queueProvider.QueueCount + " Buffor: " + queueProvider.BufforCount);
}
else
{
Console.WriteLine("Not Enqueue: " + doc.Id + "taskId: 1");
}
}
};
timer1.Enabled = true;
timer1.Start();
});
tasks.Add(task1);
var task2 = Task.Factory.StartNew(() =>
{
var timer1 = new Timer(1000) { Interval = 5000 };
timer1.Elapsed += (object sender, ElapsedEventArgs e) =>
{
foreach (var doc in docs)
{
if (queueProvider.TryEnqueue(doc))
{
Console.WriteLine("Enqueue: " + doc.Id + "taskId: 2");
Console.WriteLine("Count: " + queueProvider.QueueCount + " Buffor: " + queueProvider.BufforCount);
}
else
{
Console.WriteLine("Not Enqueue: " + doc.Id + "taskId: 2");
}
}
};
timer1.Enabled = true;
timer1.Start();
});
tasks.Add(task2);
Task.WaitAll(tasks.ToArray());
}
catch (Exception e)
{
Console.WriteLine(e);
}
Console.ReadKey();
}
}
public interface IItem
{
Guid Id { get; set; }
}
public interface IProvider
{
}
public class Provider: IProvider
{
private readonly Queue<IItem> queue;
private readonly Dictionary<Guid, DateTime> inputBuffor;
private readonly object locker = new object();
private readonly int maxQueueCount = 3;
public Provider()
{
queue = new Queue<IItem>();
inputBuffor = new Dictionary<Guid, DateTime>();
}
public bool TryEnqueue(IItem feedingItem)
{
lock (locker)
{
if (inputBuffor.ContainsKey(feedingItem.Id) || queue.Count >= maxQueueCount) return false;
inputBuffor.Add(feedingItem.Id, DateTime.Now);
queue.Enqueue(feedingItem);
return true;
}
}
public IItem Dequeue()
{
lock (locker)
{
if (queue.Count <= 0) return null;
var item = queue.Dequeue();
return item;
}
}
public int QueueCount
{
get { lock (locker) return queue.Count; }
}
public int BufforCount
{
get { lock (locker) return inputBuffor.Count; }
}
}