UniRx observable wrappers for Entitas entity and group events

ChangeType.cs

using System;

namespace Entitas {

    /** Used in the *AnyChangeObservable methods to observe multiple change types */
    [Flags]
    public enum ChangeType : short{
        Addition = 1 << 0,
        Replacement = 1 << 1,
        Removal = 1 << 2,  
    };
}

EntityUniRxAdditions.cs

using System;
using UniRx;

namespace Entitas {

    public class ComponentReplacedEventArgs<T> : EventArgs {
        public Entity entity;
        public int index;
        public T previous;
        public T current;
    }

    public class EntityChangedEventArgs<T> : EventArgs {
        public Entity entity;
        public int index;
        public T component;
    }

    public partial class Entity {

        // Component replaced observable
        public IObservable<ComponentReplacedEventArgs<T>> OnComponentReplacedAsObservable<T>(){
            var componentReplaced = Observable.FromEvent<ComponentReplaced, ComponentReplacedEventArgs<T>>(handler => {
                ComponentReplaced componentReplacedHandler = (entity, index, previousComponent, newComponent) => {
                    if(previousComponent is T){
                        var args = new ComponentReplacedEventArgs<T>();
                        args.entity = entity;
                        args.index = index;
                        args.previous = (T)previousComponent;
                        args.current = (T)newComponent;
                        handler(args);
                    }
                };
                return componentReplacedHandler; 
            },
            componentReplacedHandler => OnComponentReplaced += componentReplacedHandler,
            componentReplacedHandler => OnComponentReplaced -= componentReplacedHandler);
            return componentReplaced;
        }

        // Component added observable
        public IObservable<EntityChangedEventArgs<T>> OnComponentAddedAsObservable<T>(){
            var componentAdded = Observable.FromEvent<EntityChanged, EntityChangedEventArgs<T>>(handler => {
                EntityChanged componentAddedHandler = (entity, index, component) => {
                    if(component is T){
                        var args = new EntityChangedEventArgs<T>();
                        args.entity = entity;
                        args.index = index;
                        args.component = (T)component;
                        handler(args);
                    }
                };
                return componentAddedHandler;
            },
            componentAddedHandler => OnComponentAdded += componentAddedHandler,
            componentAddedHandler => OnComponentAdded -= componentAddedHandler);
            return componentAdded;
        }

        // Component removed observable
        public IObservable<EntityChangedEventArgs<T>> OnComponentRemovedAsObservable<T>(){
            var componentRemoved = Observable.FromEvent<EntityChanged, EntityChangedEventArgs<T>>(handler => {
                EntityChanged componentRemovedHandler = (entity, index, component) => {
                    if(component is T){
                        var args = new EntityChangedEventArgs<T>();
                        args.entity = entity;
                        args.index = index;
                        args.component = (T)component;
                        handler(args);
                    }
                };
                return componentRemovedHandler; 
            },
            componentRemovedHandler => OnComponentRemoved += componentRemovedHandler,
            componentRemovedHandler => OnComponentRemoved -= componentRemovedHandler);
            return componentRemoved;
        }
    }
}

EntityUniRxExtendedAdditions.cs

using System;
using System.Collections.Generic;
using UniRx;

namespace Entitas {
    
    public class AnyComponentChangeEventArgs<T> : EventArgs {
        public EntityChangedEventArgs<T> addition;
        public ComponentReplacedEventArgs<T> replacement;
        public EntityChangedEventArgs<T> removal;
    }

    public partial class Entity {

        /** A component of type T had any of the changes specified in 'changeTypes'. The appropriate 'replacement', 'addition' or 'removal' field will be set in the resulting AnyComponentChangeEventArgs. */

        public IObservable<AnyComponentChangeEventArgs<T>> OnAnyChangeObservable<T>(ChangeType changeTypes = ChangeType.Addition | ChangeType.Replacement | ChangeType.Removal){

            List<IObservable<AnyComponentChangeEventArgs<T>>> observedChanges = new List<IObservable<AnyComponentChangeEventArgs<T>>>();

            if((changeTypes & ChangeType.Addition) == ChangeType.Addition){
                var componentAdded  = OnComponentAddedAsObservable<T>().Select<EntityChangedEventArgs<T>, AnyComponentChangeEventArgs<T>>(addition => {
                    var change = new AnyComponentChangeEventArgs<T>();
                    change.addition = addition;
                    return change;
                });            
                observedChanges.Add(componentAdded);
            }

            if((changeTypes & ChangeType.Replacement) == ChangeType.Replacement){
                var componentReplaced = OnComponentReplacedAsObservable<T>().Select<ComponentReplacedEventArgs<T>, AnyComponentChangeEventArgs<T>>(replacement => {
                    var change = new AnyComponentChangeEventArgs<T>();
                    change.replacement = replacement;
                    return change;
                });
                observedChanges.Add(componentReplaced);
            }    

            if((changeTypes & ChangeType.Removal) == ChangeType.Removal){
                var componentRemoved  = OnComponentRemovedAsObservable<T>().Select<EntityChangedEventArgs<T>, AnyComponentChangeEventArgs<T>>(removal => {
                    var change = new AnyComponentChangeEventArgs<T>();
                    change.removal = removal;
                    return change;
                });            
                observedChanges.Add(componentRemoved);
            }

            return Observable.Merge(observedChanges);
        }
    }
}

GroupUniRxAdditions.cs

using System;
using UniRx;

namespace Entitas {
    
    public class GroupChangedEventArgs<T> : EventArgs {
        public Group group;
        public Entity entity;
        public int index;
        public T component;
    }

    public class GroupUpdatedEventArgs<T> : EventArgs {
        public Group group;
        public Entity entity;
        public int index;
        public T previous;
        public T current;
    }

    public partial class Group {

        // Entity added observable
        public IObservable<GroupChangedEventArgs<T>> OnEntityAddedAsObservable<T>(){
            var entityAdded = Observable.FromEvent<GroupChanged, GroupChangedEventArgs<T>>(handler => {
                GroupChanged entityAddedHandler = (group, entity, index, component) => {
                    if(component is T){
                        var args = new GroupChangedEventArgs<T>();
                        args.group = group;
                        args.entity = entity;
                        args.index = index;
                        args.component = (T)component;
                        handler(args);
                    }
                };
                return entityAddedHandler; 
            },
            entityAddedHandler => OnEntityAdded += entityAddedHandler,
            entityAddedHandler => OnEntityAdded -= entityAddedHandler);
            return entityAdded;
        }

        // Entity removed observable
        public IObservable<GroupChangedEventArgs<T>> OnEntityRemovedAsObservable<T>(){
            var entityRemoved = Observable.FromEvent<GroupChanged, GroupChangedEventArgs<T>>(handler => {
                GroupChanged entityRemovedHandler = (group, entity, index, component) => {
                    if(component is T){
                        var args = new GroupChangedEventArgs<T>();
                        args.group = group;
                        args.entity = entity;
                        args.index = index;
                        args.component = (T)component;
                        handler(args);
                    }
                };
                return entityRemovedHandler; 
            },
            entityRemovedHandler => OnEntityRemoved += entityRemovedHandler,
            entityRemovedHandler => OnEntityRemoved -= entityRemovedHandler);
            return entityRemoved;
        }

        // Entity updated observable
        public IObservable<GroupUpdatedEventArgs<T>> OnEntityUpdatedAsObservable<T>(){
            var entityUpdated = Observable.FromEvent<GroupUpdated, GroupUpdatedEventArgs<T>>(handler => {
                GroupUpdated entityRemovedHandler = (group, entity, index, previous, current) => {
                    if(previous is T){
                        var args = new GroupUpdatedEventArgs<T>();
                        args.group = group;
                        args.entity = entity;
                        args.index = index;
                        args.previous = (T)previous;
                        args.current = (T)current;
                        handler(args);
                    }
                };
                return entityRemovedHandler; 
            },
            entityUpdatedHandler => OnEntityUpdated += entityUpdatedHandler,
            entityUpdatedHandler => OnEntityUpdated -= entityUpdatedHandler);
            return entityUpdated;
        }
    }
}

GroupUniRxExtendedAdditions.cs

using System;
using System.Collections.Generic;
using UniRx;

namespace Entitas {

    public class AnyEntityChangeEventArgs<T> : EventArgs {
        public GroupChangedEventArgs<T> addition;
        public GroupUpdatedEventArgs<T> update;
        public GroupChangedEventArgs<T> removal;
    }

    public partial class Group {

        /** A component of type T had any of the changes specified in 'changeTypes' within the group. The appropriate 'update', 'addition' or 'removal' field will be set in the resulting AnyEntityChangeEventArgs. */

        public IObservable<AnyEntityChangeEventArgs<T>> OnAnyEntityChangeObservable<T>(ChangeType changeTypes = ChangeType.Addition | ChangeType.Replacement | ChangeType.Removal){

            List<IObservable<AnyEntityChangeEventArgs<T>>> observedChanges = new List<IObservable<AnyEntityChangeEventArgs<T>>>();

            if((changeTypes & ChangeType.Addition) == ChangeType.Addition){
                var entityAdded = OnEntityAddedAsObservable<T>().Select<GroupChangedEventArgs<T>, AnyEntityChangeEventArgs<T>>(addition => {
                    var change = new AnyEntityChangeEventArgs<T>();
                    change.addition = addition;
                    return change;
                });
                observedChanges.Add(entityAdded);
            }

            if((changeTypes & ChangeType.Replacement) == ChangeType.Replacement){
                var entityUpdated = OnEntityUpdatedAsObservable<T>().Select<GroupUpdatedEventArgs<T>, AnyEntityChangeEventArgs<T>>(update => {
                    var change = new AnyEntityChangeEventArgs<T>();
                    change.update = update;
                    return change;
                });
                observedChanges.Add(entityUpdated);
            }       

            if((changeTypes & ChangeType.Removal) == ChangeType.Removal){
                var entityRemoved  = OnEntityRemovedAsObservable<T>().Select<GroupChangedEventArgs<T>, AnyEntityChangeEventArgs<T>>(removal => {
                    var change = new AnyEntityChangeEventArgs<T>();
                    change.removal = removal;
                    return change;
                });
                observedChanges.Add(entityRemoved);
            }

            return Observable.Merge(observedChanges);
        }
    }
}

Looks useful. Can you elaborate a bit on what benefits this brings? Entitas is already "reactive" as far as I know.

@zyzxdef Entitas already provides reactive systems, group matchers and event handlers for reacting to events in your pool. UniRx mainly provides more options on how to react within systems - and these "AsObservable" extensions for UniRx are just building blocks for richer reactions within systems.

It also makes implementing some reactions to the pool easier and more intuitive to follow. For example, I can say "start a timer when a component is added and remove it when the component is changed in any way" in a single reactive expression (this way, my system reacting to component additions doesn't have to worry about the previous state of the entities). Here's what I mean:

    public class SteppedMovementSystem : IReactiveSystem, IEnsureComponents, ISetPool {

        // React to additions of the SteppedMover....
        public TriggerOnEvent trigger {  get {  return Matcher.AllOf(Matcher.SteppedMover).OnEntityAdded(); } }

    ...
    ...
    ...

        public void Execute(List<Entity> entities) {    
            foreach (var e in entities) {
                StartStepper(e);
            }
        }

        // Create a timer. It will be removed if the SteppedMover component is changed in any way. (e.g. you change direction of the stepped mover)

        private void StartStepper(Entity e){

            Observable.Timer(new DateTimeOffset(), new TimeSpan(0,0,1))
                .TakeUntil(e.OnAnyChangeObservable<SteppedMover>())
                .Subscribe(timerCount => {
                    var position = e.position;
                    var deltaAmount = 1.0f;
                    Vector3 delta = Vector3.zero;
                    switch(e.steppedMover.direction){
                        case SteppedMover.Direction.Left:
                        delta = new Vector3(deltaAmount, 0, 0);
                        break;
                        case SteppedMover.Direction.Right:
                        delta = new Vector3(-deltaAmount, 0, 0);
                        break;
                        case SteppedMover.Direction.Up:
                        delta = new Vector3(0, deltaAmount, 0);
                        break;
                        case SteppedMover.Direction.Down:
                        delta = new Vector3(0, -deltaAmount, 0);
                        break;
                    }
                    e.ReplacePosition(position.x + delta.x, 
                                      position.y + delta.y, 
                                      position.z + delta.z);

                }).AddTo(e.view.gameObject);
        }
    }
...
...

With this, you also have the option of creating "Initialize" systems (instead of reactive ones) and match entities with richer expressions than what TriggerOnEvent allows you to match (e.g. match entities, but ignore certain ones, etc).

Also, with Rx in general, you have more tools in your disposal for creating interesting signals to respond to - like filtering out certain events, terminating observations after certain events, only reacting after accumulating various different signals that you want, etc.

Thanks for your detailed explanation!

Edit: maybe you can contact the Entitas guys so they can link to this gist in their wiki.

Moved to a repo https://github.com/ancientbuho/entitas-unirx-additions