//
//  RxRealmWriteSinks.swift
//  RxRealm
//
//  Created by Marin Todorov on 6/4/16.
//  Copyright © 2016 CocoaPods. All rights reserved.
//

import RealmSwift
import RxBlocking
import RxCocoa
import RxRealmDJB
import RxSwift
import XCTest

enum WriteError: Error {
    case def
}

class RxRealmWriteSinks: XCTestCase {
    func testRxAddObjectWithSuccess() {
        let realm = realmInMemory(#function)
        let items = Observable.array(from: realm.objects(Message.self))
            .map { $0.map { $0.text } }

        // show all speakers
        DispatchQueue.main.async {
            _ = Observable.just(Message("1")).subscribe(realm.rx.add())
        }

        let result = try! items.skip(1).toBlocking(timeout: 1).first()!
        XCTAssertEqual(result[0], "1")
    }

    func testRxAddObjectWithError() {
        var conf = Realm.Configuration()
        conf.fileURL = URL(string: "/asdasdasdsad")!

        let recordedError = BehaviorRelay<Error?>(value: nil)

        DispatchQueue.main.async {
            _ = Observable.just(Message("0"))
                .subscribe(Realm.rx.add(configuration: conf, update: .all, onError: { _, error in
                    recordedError.accept(error)
                }))
        }

        let error = try! recordedError.asObservable().skip(1).toBlocking(timeout: 1).first()!
        XCTAssertNotNil(error)
        XCTAssertEqual((error! as NSError).code, 3)
    }

    func testRxAddSequenceWithSuccess() {
        let realm = realmInMemory(#function)
        let items = Observable.array(from: realm.objects(Message.self))
            .map { $0.map { $0.text } }

        // show all speakers
        DispatchQueue.main.async {
            _ = Observable.just([Message("1"), Message("2")]).subscribe(realm.rx.add())
        }

        let result = try! items.skip(1).toBlocking(timeout: 1).first()!
        XCTAssertEqual(result[0], "1")
        XCTAssertEqual(result[1], "2")
    }

    func testRxAddSequenceWithError() {
        var conf = Realm.Configuration()
        conf.fileURL = URL(string: "/asdasdasdsad")!

        let recordedError = BehaviorRelay<Error?>(value: nil)

        // show all speakers
        DispatchQueue.main.async {
            _ = Observable.from([Message("1"), Message("2")])
                .subscribe(Realm.rx.add(configuration: conf, update: .all, onError: { _, error in
                    recordedError.accept(error)
                }))
        }

        let error = try! recordedError.asObservable().skip(1).toBlocking(timeout: 1).first()!
        XCTAssertNotNil(error)
        XCTAssertEqual((error! as NSError).code, 3)
    }

    func testRxAddUpdateObjectsWithSucess() {
        let realm = realmInMemory(#function)
        let items = Observable.array(from: realm.objects(UniqueObject.self).sorted(byKeyPath: "id"))
            .map { $0.map { $0.id } }

        // show all speakers
        DispatchQueue.main.async {
            _ = Observable.just([UniqueObject(1), UniqueObject(2)]).subscribe(realm.rx.add())
            _ = Observable.just([UniqueObject(1), UniqueObject(3)]).subscribe(realm.rx.add(update: .all))
        }

        let result = try! items.skip(1).take(2).toBlocking(timeout: 1).toArray()
        XCTAssertEqual(result[0], [1, 2])
        XCTAssertEqual(result[1], [1, 2, 3])
    }

    func testRxDeleteItem() {
        let realm = realmInMemory(#function)
        let items = Observable.array(from: realm.objects(UniqueObject.self))
            .map { $0.map { $0.id } }

        let object1 = UniqueObject(1)
        let object2 = UniqueObject(2)
        try! realm.write {
            realm.add([object1, object2])
        }

        // show all speakers
        DispatchQueue.main.async {
            _ = Observable.just(object1).subscribe(realm.rx.delete())
            _ = Observable.just(object2).subscribe(Realm.rx.delete())
        }

        let result = try! items.take(3).toBlocking(timeout: 1).toArray()
        XCTAssertEqual(result[0], [1, 2])
        XCTAssertEqual(result[1], [2])
        XCTAssertEqual(result[2], [])
    }

    func testRxDeleteItemWithError() {
        let object1 = UniqueObject(1)
        let recordedError = BehaviorRelay<Error?>(value: nil)

        DispatchQueue.main.async {
            _ = Observable.just(object1)
                .subscribe(Realm.rx.delete(onError: { _, error in
                    recordedError.accept(error)
                }))
        }

        let error = try! recordedError.asObservable().skip(1).toBlocking(timeout: 1).first()!
        XCTAssertNotNil(error)
    }

    func testRxDeleteItemsWithSuccess() {
        let realm = realmInMemory(#function)
        let items = Observable.array(from: realm.objects(UniqueObject.self).sorted(byKeyPath: "id"))
            .map { $0.map { $0.id } }

        let object1 = UniqueObject(1)
        let object2 = UniqueObject(2)
        let object3 = UniqueObject(3)
        let object4 = UniqueObject(4)

        try! realm.write {
            realm.add([object1, object2, object3, object4])
        }

        // show all speakers
        DispatchQueue.main.async {
            _ = Observable.just([object1, object2]).subscribe(realm.rx.delete())
            _ = Observable.just([object3, object4]).subscribe(Realm.rx.delete())
        }

        let result = try! items.take(3).toBlocking(timeout: 1).toArray()
        XCTAssertEqual(result[0], [1, 2, 3, 4])
        XCTAssertEqual(result[1], [3, 4])
        XCTAssertEqual(result[2], [])
    }

    func testRxDeleteItemsWithError() {
        let object1 = UniqueObject(1)
        let object2 = UniqueObject(2)
        let recordedError = BehaviorRelay<Error?>(value: nil)

        DispatchQueue.main.async {
            _ = Observable.just([object1, object2])
                .subscribe(Realm.rx.delete(onError: { _, error in
                    recordedError.accept(error)
                }))
        }

        let error = try! recordedError.asObservable().skip(1).toBlocking(timeout: 1).first()!
        XCTAssertNotNil(error)
    }

    func testRxAddObjectsFromDifferentThreads() {
        let realm = realmInMemory(#function)
        let conf = realm.configuration

        let items = Observable.array(from: realm.objects(UniqueObject.self).sorted(byKeyPath: "id"))
            .map { $0.map { $0.id } }

        // write on current thread
        _ = Observable.just(UniqueObject(1)).subscribe(realm.rx.add())

        // write on background thread
        DispatchQueue.global(qos: .background).async {
            let realm = try! Realm(configuration: conf)
            _ = Observable.just(UniqueObject(2))
                .subscribe(realm.rx.add())
        }

        // write on main scheduler
        DispatchQueue.global(qos: .background).async {
            _ = Observable.just(UniqueObject(3))
                .observeOn(MainScheduler.instance)
                .subscribe(Realm.rx.add(configuration: conf))
        }

        // write on bg scheduler
        DispatchQueue.main.async {
            _ = Observable.just(UniqueObject(4))
                .observeOn(ConcurrentDispatchQueueScheduler(queue: DispatchQueue.global(qos: .background)))
                .subscribe(Realm.rx.add(configuration: conf))
        }

        // subscribe on main, write in bg
        DispatchQueue.main.async {
            _ = Observable.just([UniqueObject(5), UniqueObject(6)])
                .observeOn(ConcurrentDispatchQueueScheduler(queue: DispatchQueue.global(qos: .background)))
                .subscribe(Realm.rx.add(configuration: conf))
        }

        let until = Observable.array(from: realm.objects(UniqueObject.self))
            .filter { $0.count == 6 }
            .delay(.milliseconds(100), scheduler: MainScheduler.instance)

        let result = try! items.takeUntil(until).toBlocking(timeout: 1).toArray()
        XCTAssertEqual(result.last!, [1, 2, 3, 4, 5, 6])
    }
}
