<commit_before>//
//  StreamBase.swift
//  StreamBaseKit
//
//  Created by <NAME> on 8/31/15.
//  Copyright (c) 2015 Movem3nt, Inc. All rights reserved.
//

import Firebase

/**
    Surfaces a Firebase collection as a stream suitable for presenting in a ui table or 
    collection.
    
    In addition to basic functionality of keeping stream synched with Firebase backing store
    and invoking a delegate with updates, it has some more advanced features:
    
    * Inverted streams
    * Paging
    * Predicates
    * Batching
*/
public class StreamBase : StreamBaseProtocol {
    public typealias Predicate = StreamBaseItem -> Bool
    public typealias Comparator = (StreamBaseItem, StreamBaseItem) -> Bool
    public typealias QueryPager = (start: AnyObject?, end: AnyObject?, limit: Int?) -> FQuery

    public enum Ordering {
        case Key
        case Child(String)
        // TODO Priority
    }
    
    private var handles = [UInt]()
    private var arrayBeforePredicate = KeyedArray<StreamBaseItem>()
    private var array = KeyedArray<StreamBaseItem>()
    private var batchArray = KeyedArray<StreamBaseItem>()
    
    private let type: StreamBaseItem.Type!
    private let query: FQuery!
    private let queryPager: QueryPager!
    private let limit: Int?
    
    private var observer: NSObjectProtocol?
    private var isBatching = false
    private var timer: NSTimer?
    private var isFetchingMore = false
    private var shouldTellDelegateInitialLoadIsDone: Bool? = false
    
    /**
        The comparator function used for sorting items.
    */
    public let comparator: Comparator
    
    /**
        How long to wait for a batch of changes to accumulate before invoking delegate.
        If nil, then there is no delay.  The unit is seconds.
    */
    public var batchDelay: NSTimeInterval? = 0.1
    
    /**
        A value used to decide whether elements are members of the stream or not.  This
        predicate is evaluated client-side, and so scalability and under-fetching must
        be taken into account.
    */
    public var predicate: Predicate? {
        didSet {
            batching { a in
                if let p = self.predicate {
                    a.clear()
                    for t in self.arrayBeforePredicate {
                        if p(t) {
                            a.append(t)
                        }
                    }
                } else {
                    a.reset(self.arrayBeforePredicate)
                }
            }
        }
    }
    
    /**
        Limit the results after predicate to this value.  If the fetch limit is combined with
        a predicate, then under-fetching may result.
    */
    public var afterPredicateLimit: Int?
    
    /**
        The delegate to notify as the underying data changes.
    */
    public weak var delegate: StreamBaseDelegate? {
        didSet {
            if limit == 0 {
                delegate?.streamDidFinishInitialLoad(nil)
            }
        }
    }

    /**
        Construct an empty stream instance, not connected to any backing Firebase store.
    */
    public init() {
        type = nil
        limit = 0
        query = nil
        queryPager = nil
        comparator = { $0.key < $1.key }
    }
    
    /**
        Construct a stream instance containing items of the given type at the given Firebase reference.

        :param: type    The type of items in the stream.
        :param: ref The Firebase reference where the items are stored.
        :param: limit   The max number to fetch.
        :param: ascending   Whether to materialize the underlying array in ascending or descending order.
        :param: ordering    The ordering to use.
    */
    public convenience init(type: StreamBaseItem.Type, ref: Firebase, limit: Int? = nil, ascending: Bool = true, ordering: Ordering = .Key) {
        let queryBuilder = QueryBuilder(ref: ref)
        queryBuilder.limit = limit
        queryBuilder.ascending = ascending
        queryBuilder.ordering = ordering
        self.init(type: type, queryBuilder: queryBuilder)
    }
    
    /**
        Construct a stream instance containing items of the given type using the query builder specification.
        
        :param: type    The type of items in the stream.
        :param: queryBuilder    The details of what firebase data to query.
    */
    public init(type: StreamBaseItem.Type, queryBuilder: QueryBuilder) {
        self.type = type
        
        limit = queryBuilder.limit
        comparator = queryBuilder.buildComparator()
        query = queryBuilder.buildQuery()
        queryPager = queryBuilder.buildQueryPager()
        
        handles.append(query.observeEventType(.ChildAdded, withBlock: { [weak self] snapshot in
            if let s = self {
                var item = s.type(key: snapshot.key, ref: snapshot.ref, dict: snapshot.value as? [String: AnyObject])
                s.arrayBeforePredicate.append(item)
                if s.predicate == nil || s.predicate!(item) {
                    s.batching { $0.append(item) }
                }
            }
            }))
        
        handles.append(query.observeEventType(.ChildRemoved, withBlock: { [weak self] snapshot in
            if let s = self {
                if let row = s.arrayBeforePredicate.find(snapshot.key) {
                    s.arrayBeforePredicate.removeAtIndex(row)
                }
                s.batching { a in
                    if let row = a.find(snapshot.key) {
                        a.removeAtIndex(row)
                    }
                }
            }
            }))
        
        handles.append(query.observeEventType(.ChildChanged, withBlock: { [weak self] snapshot in
            if let s = self {
                if let row = s.arrayBeforePredicate.find(snapshot.key) {
                    let t = s.arrayBeforePredicate[row]
                    t.update(snapshot.value as! [String: AnyObject])
                    s.handleItemChanged(t)
                }
            }
            }))
        
        var inflight: Inflight? = Inflight()
        query.observeSingleEventOfType(.Value, withBlock: { [weak self] snapshot in
            if let s = self {
                inflight = nil
                s.shouldTellDelegateInitialLoadIsDone = true
                s.scheduleBatch()
            }
            }, withCancelBlock: { [weak self] error in
                self?.delegate?.streamDidFinishInitialLoad(error)
            })
    }
    
    deinit {
        if let obs = self.observer {
            NSNotificationCenter.defaultCenter().removeObserver(obs)
        }
        for h in handles {
            query.removeObserverWithHandle(h)
        }
    }
    
    /**
        Find the item for a given key.
    
        :param: key The key to check
        :returns:    The item or nil if not found.
    */
    public func find(key: String) -> StreamBaseItem? {
        if let row = array.find(key) {
            return array[row]
        }
        return nil
    }
    
    /**
        Find the index path for the given key.
<commit_msg>no longer need observer<commit_after>//
//  StreamBase.swift
//  StreamBaseKit
//
//  Created by <NAME> on 8/31/15.
//  Copyright (c) 2015 Movem3nt, Inc. All rights reserved.
//

import Firebase

/**
    Surfaces a Firebase collection as a stream suitable for presenting in a ui table or 
    collection.
    
    In addition to basic functionality of keeping stream synched with Firebase backing store
    and invoking a delegate with updates, it has some more advanced features:
    
    * Inverted streams
    * Paging
    * Predicates
    * Batching
*/
public class StreamBase : StreamBaseProtocol {
    public typealias Predicate = StreamBaseItem -> Bool
    public typealias Comparator = (StreamBaseItem, StreamBaseItem) -> Bool
    public typealias QueryPager = (start: AnyObject?, end: AnyObject?, limit: Int?) -> FQuery

    public enum Ordering {
        case Key
        case Child(String)
        // TODO Priority
    }
    
    private var handles = [UInt]()
    private var arrayBeforePredicate = KeyedArray<StreamBaseItem>()
    private var array = KeyedArray<StreamBaseItem>()
    private var batchArray = KeyedArray<StreamBaseItem>()
    
    private let type: StreamBaseItem.Type!
    private let query: FQuery!
    private let queryPager: QueryPager!
    private let limit: Int?
    
    private var isBatching = false
    private var timer: NSTimer?
    private var isFetchingMore = false
    private var shouldTellDelegateInitialLoadIsDone: Bool? = false
    
    /**
        The comparator function used for sorting items.
    */
    public let comparator: Comparator
    
    /**
        How long to wait for a batch of changes to accumulate before invoking delegate.
        If nil, then there is no delay.  The unit is seconds.
    */
    public var batchDelay: NSTimeInterval? = 0.1
    
    /**
        A value used to decide whether elements are members of the stream or not.  This
        predicate is evaluated client-side, and so scalability and under-fetching must
        be taken into account.
    */
    public var predicate: Predicate? {
        didSet {
            batching { a in
                if let p = self.predicate {
                    a.clear()
                    for t in self.arrayBeforePredicate {
                        if p(t) {
                            a.append(t)
                        }
                    }
                } else {
                    a.reset(self.arrayBeforePredicate)
                }
            }
        }
    }
    
    /**
        Limit the results after predicate to this value.  If the fetch limit is combined with
        a predicate, then under-fetching may result.
    */
    public var afterPredicateLimit: Int?
    
    /**
        The delegate to notify as the underying data changes.
    */
    public weak var delegate: StreamBaseDelegate? {
        didSet {
            if limit == 0 {
                delegate?.streamDidFinishInitialLoad(nil)
            }
        }
    }

    /**
        Construct an empty stream instance, not connected to any backing Firebase store.
    */
    public init() {
        type = nil
        limit = 0
        query = nil
        queryPager = nil
        comparator = { $0.key < $1.key }
    }
    
    /**
        Construct a stream instance containing items of the given type at the given Firebase reference.

        :param: type    The type of items in the stream.
        :param: ref The Firebase reference where the items are stored.
        :param: limit   The max number to fetch.
        :param: ascending   Whether to materialize the underlying array in ascending or descending order.
        :param: ordering    The ordering to use.
    */
    public convenience init(type: StreamBaseItem.Type, ref: Firebase, limit: Int? = nil, ascending: Bool = true, ordering: Ordering = .Key) {
        let queryBuilder = QueryBuilder(ref: ref)
        queryBuilder.limit = limit
        queryBuilder.ascending = ascending
        queryBuilder.ordering = ordering
        self.init(type: type, queryBuilder: queryBuilder)
    }
    
    /**
        Construct a stream instance containing items of the given type using the query builder specification.
        
        :param: type    The type of items in the stream.
        :param: queryBuilder    The details of what firebase data to query.
    */
    public init(type: StreamBaseItem.Type, queryBuilder: QueryBuilder) {
        self.type = type
        
        limit = queryBuilder.limit
        comparator = queryBuilder.buildComparator()
        query = queryBuilder.buildQuery()
        queryPager = queryBuilder.buildQueryPager()
        
        handles.append(query.observeEventType(.ChildAdded, withBlock: { [weak self] snapshot in
            if let s = self {
                var item = s.type(key: snapshot.key, ref: snapshot.ref, dict: snapshot.value as? [String: AnyObject])
                s.arrayBeforePredicate.append(item)
                if s.predicate == nil || s.predicate!(item) {
                    s.batching { $0.append(item) }
                }
            }
            }))
        
        handles.append(query.observeEventType(.ChildRemoved, withBlock: { [weak self] snapshot in
            if let s = self {
                if let row = s.arrayBeforePredicate.find(snapshot.key) {
                    s.arrayBeforePredicate.removeAtIndex(row)
                }
                s.batching { a in
                    if let row = a.find(snapshot.key) {
                        a.removeAtIndex(row)
                    }
                }
            }
            }))
        
        handles.append(query.observeEventType(.ChildChanged, withBlock: { [weak self] snapshot in
            if let s = self {
                if let row = s.arrayBeforePredicate.find(snapshot.key) {
                    let t = s.arrayBeforePredicate[row]
                    t.update(snapshot.value as! [String: AnyObject])
                    s.handleItemChanged(t)
                }
            }
            }))
        
        var inflight: Inflight? = Inflight()
        query.observeSingleEventOfType(.Value, withBlock: { [weak self] snapshot in
            if let s = self {
                inflight = nil
                s.shouldTellDelegateInitialLoadIsDone = true
                s.scheduleBatch()
            }
            }, withCancelBlock: { [weak self] error in
                self?.delegate?.streamDidFinishInitialLoad(error)
            })
    }
    
    deinit {
        for h in handles {
            query.removeObserverWithHandle(h)
        }
    }
    
    /**
        Find the item for a given key.
    
        :param: key The key to check
        :returns:    The item or nil if not found.
    */
    public func find(key: String) -> StreamBaseItem? {
        if let row = array.find(key) {
            return array[row]
        }
        return nil
    }
    
    /**
        Find the index path for the given key.

{'fraction_non_alphanumeric': 0.07516759378120097, 'fraction_numerical': 0.0024247610897161604, 'mean_word_length': 1.7024474850645597, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6862329', 'n_tokens_mistral': 3716, 'n_tokens_neox': 3570, 'n_words': 1196}