 
      - returns: `Observable<(Self, RealmChangeset?)>`
      */
-    public static func changesetFrom(_ collection: E, scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<(E, RealmChangeset?)> {
+    public static func changesetFrom(_ collection: E, scheduler: ImmediateSchedulerType = CurrentThreadScheduler.instance) -> Observable<(AnyRealmCollection<E.ElementType>, RealmChangeset?)> {
         return Observable.create {observer in
-            let token = collection.addNotificationBlock {changeset in
+            let token = collection.toAnyCollection().addNotificationBlock {changeset in
 
                 switch changeset {
                     case .initial(let value):

*/
public protocol NotificationEmitter {

    associatedtype ElementType

    /**
     Returns a `NotificationToken`, which while retained enables change notifications for the current collection.

    func addNotificationBlock(_ block: @escaping (RealmCollectionChange<Self>) -> ()) -> NotificationToken

    func toArray() -> [ElementType]

    func toAnyCollection() -> AnyRealmCollection<ElementType>
}

extension List: NotificationEmitter {
    public func toAnyCollection() -> AnyRealmCollection<T> {
        return AnyRealmCollection<ElementType>(self)
    }
    public typealias ElementType = Element
    public func toArray() -> [Element] {
        return Array(self)
