improve publisher / consumer code

Svelto.ECS/EnginesRoot.DoubleBufferedEntityViews.cs

@@ -1,4 +1,4 @@
-using Svelto.DataStructures.Experimental;
+using Svelto.DataStructures.Experimental;
 using EntitiesDB =
     Svelto.DataStructures.Experimental.FasterDictionary<uint, System.Collections.Generic.Dictionary<System.Type,
         Svelto.ECS.Internal.ITypeSafeDictionary>>;

Svelto.ECS/EnginesRoot.Entities.cs

@@ -293,6 +293,12 @@ namespace Svelto.ECS
             return _entitiesStream.GenerateConsumer<T>(name, capacity);
         }
 
+        
+        public Consumer<T> GenerateConsumer<T>(ExclusiveGroup @group, string name, int capacity) where T : unmanaged, 
+            IEntityStruct
+        {
+            return _entitiesStream.GenerateConsumer<T>(@group, name, capacity);
+        }
         const string INVALID_DYNAMIC_DESCRIPTOR_ERROR =
             "Found an entity requesting an invalid dynamic descriptor, this " +
             "can happen only if you are building different entities with the " +

Svelto.ECS/EntityStream.cs

@@ -25,6 +25,13 @@ namespace Svelto.ECS
 
             return (_streams[typeof(T)] as EntityStream<T>).GenerateConsumer(name, capacity);
         }
+        
+        public Consumer<T> GenerateConsumer<T>(ExclusiveGroup @group, string name, int capacity) where T : unmanaged, IEntityStruct
+        {
+            if (_streams.ContainsKey(typeof(T)) == false) _streams[typeof(T)] = new EntityStream<T>();
+
+            return (_streams[typeof(T)] as EntityStream<T>).GenerateConsumer( @group, name, capacity);
+        }
 
         internal void PublishEntity<T>(ref T entity) where T : unmanaged, IEntityStruct
         {
@@ -45,53 +52,66 @@ namespace Svelto.ECS
     {
         public void PublishEntity(ref T entity)
         {
-            for (int i = 0; i < _buffers.Count; i++)
-                _buffers[i].Enqueue(ref entity);
+            for (int i = 0; i < _consumers.Count; i++)
+            {
+                if (_consumers[i]._hasGroup)
+                {
+                    if (EntityBuilder<T>.HAS_EGID && (entity as INeedEGID).ID.groupID == _consumers[i]._group)
+                    {
+                        _consumers[i].Enqueue(ref entity);
+                    }
+                }
+                else
+                    _consumers[i].Enqueue(ref entity);
+            }
         }
 
         public Consumer<T> GenerateConsumer(string name, int capacity)
         {
             var consumer = new Consumer<T>(name, capacity, this);
-            _buffers.Add(consumer);
+            
+            _consumers.Add(consumer);
+            
+            return consumer;
+        }
+        
+        public Consumer<T> GenerateConsumer(ExclusiveGroup @group, string name, int capacity)
+        {
+            var consumer = new Consumer<T>(group, name, capacity, this);
+            
+            _consumers.Add(consumer);
+            
             return consumer;
         }
 
         public void RemoveConsumer(Consumer<T> consumer)
         {
-            _buffers.UnorderedRemove(consumer);
+            _consumers.UnorderedRemove(consumer);
         }
 
-        readonly FasterListThreadSafe<Consumer<T>> _buffers = new FasterListThreadSafe<Consumer<T>>();
+        readonly FasterListThreadSafe<Consumer<T>> _consumers = new FasterListThreadSafe<Consumer<T>>();
     }
 
     public struct Consumer<T>: IDisposable where T:unmanaged, IEntityStruct
     {
-        internal Consumer(string name, int capacity, EntityStream<T> stream)
+        internal Consumer(string name, int capacity, EntityStream<T> stream): this()
         {
             _ringBuffer = new RingBuffer<T>(capacity);
             _name = name;
             _stream = stream;
         }
+        
+        internal Consumer(ExclusiveGroup @group, string name, int capacity, EntityStream<T> stream):this(name, capacity, stream)
+        {
+            _group = group;
+            _hasGroup = true;
+        }
 
         internal void Enqueue(ref T entity)
         {
             _ringBuffer.Enqueue(ref entity, _name);
         }
         
-        public bool TryDequeue(ExclusiveGroup group, out T entity)
-        {
-            if (_ringBuffer.TryDequeue(out entity, _name) == true)
-            {
-                if (EntityBuilder<T>.HAS_EGID)
-                    return (entity as INeedEGID).ID.groupID == @group;
-                
-                throw new ECSException(
-                    "When an exclusive group is defined, TryDeque must operate on an entity with EGID");
-            }
-
-            return false;
-        }
-
         public bool TryDequeue(out T entity) { return _ringBuffer.TryDequeue(out entity, _name); }
         public void Flush() { _ringBuffer.Reset(); }
         public void Dispose() { _stream.RemoveConsumer(this); }
@@ -99,6 +119,7 @@ namespace Svelto.ECS
         readonly RingBuffer<T>   _ringBuffer;
         readonly EntityStream<T> _stream;
         readonly string          _name;
-
+        internal readonly ExclusiveGroup  _group;
+        internal readonly bool _hasGroup;
     }
 }

Svelto.ECS/GenericentityStreamConsumerFactory.cs

@@ -12,11 +12,18 @@ namespace Svelto.ECS
             return _enginesRoot.Target.GenerateConsumer<T>(name, capacity);
         }
 
+        public Consumer<T> GenerateConsumer<T>(ExclusiveGroup group, string name, int capacity) where T : unmanaged, IEntityStruct
+        {
+            return _enginesRoot.Target.GenerateConsumer<T>(group, name, capacity);
+        }
+
         readonly DataStructures.WeakReference<EnginesRoot> _enginesRoot;
     }
     
     public interface IEntityStreamConsumerFactory
     {
         Consumer<T> GenerateConsumer<T>(string name, int capacity) where T : unmanaged, IEntityStruct;
+        Consumer<T> GenerateConsumer<T>(ExclusiveGroup group, string name, int capacity) 
+            where T : unmanaged, IEntityStruct;
     }
 }