jensravens / interstellar Goto Github PK

I'm trying to use Queue.background and Queue.main to do some background work on another thread and then publish the result on the main thread. Based on the signatures of the Queue functions and my experience with 1.0 I'm assuming I'm supposed to use it something like this:

func longRunningTask(arg: String) -> Observable<Result<[String]>> {
    // takes a few seconds to complete, 
    return Observable(.success(["foo", "bar"])
}

func fetchArray(arg: String) -> Observable<Result<[String]>> {
    return Observable<String>(arg)
    .flatMap(Queue.background)
    .flatMap(longRunningTask)
    .flatMap(Queue.main)
}

@IBAction onClick(_ sender: Any) {
   activityIndicator.startAnimating()
   fetchArray("exampleValue")
   .next {
        activityIndicator.stopAnimating()
    }
}

Is the basically how I should be using the Queue functions?

If it is, I've found that roughly 50% of the time, activityIndicator.stopAnimating() won't be called on the main thread, but instead will be called from the background thread.

I think this is caused by the Observable created by the block created by Queue.queue<T>(_ queue: DispatchQueue) having it's update method called on the new thread before anything is subscribed to it. Then when the original thread gets around to calling subscribe the observable immediately executes the subscribe block because of the saved value update the was dispatched on the other thread.

During migration to Interstellar 2 I noticed that Thread class is not "supported" both in Signal and Observable. I tried to used it as suggested in description (Signal.ensure(Thread.main)) but ensure method expects something else. Do you plan to remove it completely or there is some other method that I didn't noticed.

It seems that after doing .map to a signal the newly created signal never gets deallocated. For example:

import Interstellar

let text = Signal<String>()

func doNothing(signal: Signal<String>) {
    let leakedSignal = signal.map { _ in
        return "I am silent"
    }.next { text in
        print(text)
    }
}

doNothing(text)

text.update("Hey")
text.update("Hey")

In this case I would expect nothing to be printed to the console because I update text signal and there are no subscribers in the current scope. However in reality I get "I am silent" two times because leakedSignal is not released after doNothing is finished.

For this particular map capturing signal weakly solves the issue:

public func map<U>(f: T -> U) -> Signal<U> {
    let signal = Signal<U>()
    subscribe { [weak signal] result in
        signal?.update(result.map(f))
    }
    return signal
}

I guess this should be the case for all variants of map and possibly other operators.

I am wrapping URLSessionDataTask instances in Observable instances via code similar to that listed below (I have removed the use of generics and replaced them with concrete types to, hopefully, make the example more clear).

// Where the conversion of the datatask to the signal occurs.
func toObservable(url: URL, dataTaskToConvert: (URL, (Result<Data>) -> Void)) -> Observable<Result<Data>> {

    let signal = Observable<Result<Data>>()

    let value: URLSessionDataTask? = dataTaskToConvert(url) { result in
        signal.update(result)
    }
    return signal
}

// Example API for consumer to make network requests without ugly callbacks.
func fetch(url: URL) ->  Observable<Result<Data>> {
    return toObservable(url: url, dataTaskToConvert: fetch)
}

// DataTask response handling is transformed to use an API with Interstellar's Result type like so:
func fetch(url: URL, completion: @escaping (Result<Data>) -> Void) -> URLSessionDataTask {
    let task = sharedUrlSession.dataTask(with: url) { data, response, error in
        if let data = data {
            completion(Result.success(data))
            return
        }

        if let error = error {
            completion(Result.error(error))
            return
        }

        let sdkError = SDKError.invalidHTTPResponse(response: response)
        completion(Result.error(sdkError))
    }

    task.resume()
    return task
}

Unfortunately, I haven't figured out a way to expose the underlying URLSessionDataTask to the client to cancel the request in a clean fashion. I think ideally, there would be somesort of method on the observable to stop the signal. My workaround currently is to return tuple of (URLSessionDataTask, Observable<Result<Data>>) rather than just the Observable so that the client can take advantage of the Observable API provided by Interstellar while also having the ability to cancel the request. However, I find the tuple API ugly to use, as client code would look like the following:

// must access the tuple element at index `1` (or named parameter)
fetch(url: someURL).1.then { data in
 // do stuff with data
}.error { 
    // handle error.
}

I am relatively new to the reactive style of programming, so I am currently unable to figure out what a good strategy for disposing the observable and it's underlying might be. I looked into the section on Disposables in the README for ReactiveKit, but I'm not sure how the same concept would integrate with Interstellar.

Is there a way you might recommend cancelling the URLSessionDataTask? I would be very happy to work on a pull request (with a small bit of guidance to kick me off in the right direction) if you think this feature might be useful.

If update is called from multiple threads at the same time, peak and the transformed value might diverge. Update should be a synchronized method.

see https://twitter.com/iceX33/status/633272192197873664

Hi,

I'm looking at using Interstellar for a nice simple signal type, but I want the .error() callback to be typed. There are two ways this could work, either the class is Signal<T,E> where E is the specific error class, or I thought without patching I could subclass specific eg:

class FooSignal : Signal<Foo> {
   public func error(g: FooError -> Void) -> signal ... etc
}

But the class is marked final. Is there any specific reason to disallow subclassing?

Thanks for publishing this great library!

I really loved it, but I quickly faced a problem (for me at least). I am trying to implement a "service-oriented architecture", which is basically just a bunch of services that are instantiated then bound together through observables to make the whole thing "tick". They should not know about each other, except for the Observable being passed in from the binding object. Doing this makes it impossible to clean up the observer, since unsubscribe requires access to the Observable.

Pondering on this a bit, I thought of this, and would like to hear your thoughts about it. It could use a bit of type-erasure, and other tune ups, but was wondering if you're interested in this direction:

final class Subscription<T> {

    private weak var observable: Observable<T>?
    private let token: ObserverToken

    init(observable: Observable<T>, token: ObserverToken) {
        self.observable = observable
        self.token = token
    }

    deinit {
        observable?.unsubscribe(token)
    }
}

extension ObserverToken {

    func subscription<T>(from observable: Observable<T>) -> Subscription<T> {
        return Subscription(observable: observable, token: self)
    }
}

Debounce seems to deliver the first skipped call to update, instead of the last call to update.

I think this happens because the skipped calls to update are scheduled to be evaluated again 1) without previous skipped calls being cancelled, and 2) are scheduled to be evaluated again very close to each other temporally. Observable.delay uses DispatchQueue.asyncAfter which isn't very precise, so whichever one fires first (usually the first one scheduled in my search-as-you-type use case) will set the new lastCalled value and ignore the others.

// skip result if there was a newer result
if currentTime.compare(lastCalled!) == .orderedDescending {
    let s = Observable<T>()
    s.delay(seconds - timeSinceLastCall!).subscribe(updateIfNeeded(observable))
    s.update(value)
}

I don't think it's possible to "silently" update an observable's value. Such a use-case kinda popped-up for me today...

Do you think it makes sense to add it to Interstellar's API?

Hi!

I'm using Interstellar in our project and so far it's working fine.
There is a use case where I can update a signal multiple times in a short period but I'm only interested in one update every let's say half a second. It's fine to discard the previous updates, I'm just interested in the latest one.
I'm using debounce for this, but looking at the tests (and at the wrong outcome in the app), calling next on a debounced signal will still fire the callback as many times as updates on the original signal.
Is this the intended behavior and how can I discard previous updates?

The main goal I'm trying to achieve is to have some kind of buffer (a [Int]) that can be filled and flushed every half a second at most

Thanks!

While trying to compile feature/v2 with Xcode 8.1, OSX 10.12 in Carthage, I get:

Carthage/Checkouts/Interstellar/Sources/Threading.swift:47:23: error: 'global(qos:)' is only available on OS X 10.10 or newer

Carthage/Checkouts/Interstellar/Sources/Threading.swift:47:36: error: 'background' is only available on OS X 10.10 or newer

A shell task failed with exit code 65:
** BUILD FAILED **

The following build commands failed:
CompileSwift normal x86_64
CompileSwiftSources normal x86_64 com.apple.xcode.tools.swift.compiler
(2 failures)

This is my first approach:

let a = Signal()
a.flatMap() {
    () -> Int in
    return 42
}
a.next {
    print($0)
}
a.update()

Unfortunately next() gets fired twice.

This issue is collecting all feature requests that require a breaking change to the public API of this library. Please discuss features in the corresponding issues (linked in the description):

• Split up Signal<T> and Result<T>. This also adds support for non-failing signals and fixes the naming of some methods (flatMap becomes then, so flatMap is free to do a real flat-mapping). #15
• Support Signal<Void>. This might require to get rid of sending the initial signal value on subscribe. #13
• Add a way to unsubscribe from a signal using a subscribtion-object. #12

Why does wait throw an error or return a value instead of returning a result? This makes it very hard to use wait while continuing to process signals.

Hi Jens,

Just a suggestion, how about refactoring debounce to GCD source, so that we can schedule and cancel when needed, like https://github.com/onmyway133/Signal/blob/master/Pod/Classes/Signal%2BDebounce.swift and http://mokagio.github.io/tech-journal/2014/11/18/objective-c-throttling-with-gcd.html

Hi,
I'm huge fan of the project. It is really well done and using it is a breeze.

Today, Apple released Xcode 10.2 with Swift 5 and now we have Result type built into the language. How do you think it influences the project? Should it replace Interstellar's Result type or would you like to keep them both by conforming to ResultType protocol?

Merging multiple signals means you end up with a "lops sided" tuple. Is there a better way to merge multiple signals?

let signal = Signal("Hello")
signal.merge(Signal("world"))
.merge(Signal("how are you?"))
.merge(Signal("42"))
.next() {
    print($0)        // ((("Hello", "world"), "how are you?"), "42")
    print($0.0.0.1)  // world
}

Missing comma in .podspec file on branch feature/v2 in line

cs.source_files = ["Sources/Result.swift", "Sources/Signal.swift", "Sources/Observable.swift", "Sources/ObserverToken.swift" "Sources/Mutex.swift", "Sources/ResultType.swift", "Sources/Observable+Result.swift"]

(no comma between "Sources/ObserverToken.swift" and "Sources/Mutex.swift")

There is no mechanism to unsubscribe from signals.
Thus over the lifetime of the application callbacks to subscription blocks referencing deallocated objects will be sent. The amount of unnecessary callbacks will steadily increase while the application is used.
This will finally lead to performance issues, especially in the case when Signals are used to update UI elements in view controllers.

Currently this library uses FRP in the sense of this article. The original definition though is a bit different (see this issue on reactive cocoa). We should implement a disclaimer in the readme to state that this is inspired by FRP and not implementing real FRP semantics.

First of all - i am loving the simplicity of interstellar!
ty for putting it online.

In my projects i added an extension for Signals.
I often use this in case of 1 Signal with a Tuple of values.

Maybe you see some use case for this.

extension Signal {
  func update(f: T? -> T){
    // internally:  self.update(f(self.value?.value))
    self.update(f(self.peek()))
  }
}

Swift2's throw is just some syntactic sugar around a result type, therefore flatMap/bind should support throwing functions/methods.

Is there a reason other than lack of resources for not including filter on observables?
I would be happy to add a pull request unless I am missing a particular design decision.

Wouldn't it be better to have the Signal object be a struct rather than a final class. This is ensures safety when copying and is preferred in Swift (as seen in this talk) over classes. Really, I just wanted to know why it was a final class.

...would be awesome.

I'm using a modified fork of
https://github.com/madhavajay/Interstellar/tree/swift-3
at the moment, but it would be grand to have something official.

And thank you for your awesome framework! 🙏

The current subscription/unsubscription mechanism in v2 relies on having both the Observable and the ObserverToken, which complicates things slightly. For use in a table view cell, for example, we need to unsubscribe during prepareForReuse(), and afterward set the observable and the token to be nil. This requires some awkwardness, either with:

if let observerToken = observerToken, observable = observable {
    observable.unsubscribe(observerToken)
}
self.observerToken = nil
self.observable = nil

Or something with tuples:

var subscription: (ObserverToken, Observable<Whatever>)?

...

if let subscription = subscription {
    subscription.1.unsubscrive(subscription.0)
}
subscription = nil

Neither is particularly awesome. I'd like to consider using a more complex ObserverToken that has a weak or unowned reference to the observable it's bound to, so we don't need to hang on to both observable and observer token. Any thoughts? Happy to send a PR if you like the idea.

Hey, I was curious on what the intended/expected minimum OS X target is.

The README says the requirements are iOS 7.0+ / Mac OS X 10.9+ / Ubuntu 14.10 but when I use Carthage fetch and install Interstellar it results in a framework with minimum deployment target of OS X 10.11

I tried to change the Interstellar.xcodeproj to set a deployment target of 10.10, it builds fine... but I'm wondering whether its just an issue with the .xcodeproj and it should work fine or if the documentation is out of date. Any guidance would be helpful, thanks.

flatMap should have a function signature of Signal a -> (a -> Signal b) -> Signal b, but it is currently Signal a -> (a -> b) -> Signal b which is a functor mapping, not flatMapping.

the swift standard library tends to go with flatMap in Swift2

In function subscribe

    public func subscribe(f: Result<T> -> Void) -> Signal<T> {
        if let value = value {
            f(value)
        }
        mutex.lock {
            callbacks.append(f)
        }
        return self
    }

https://github.com/JensRavens/Interstellar/blob/master/Sources/Signal.swift#L145-L153

it seems to me, there's a potential data race: access to value should be protected by the mutex.

Proposed solution:

    public func subscribe(f: Result<T> -> Void) -> Signal<T> {
        var v: Result<T>?
        mutex.lock {
            v = self.value
            callbacks.append(f)
        }
        if let value = v {
            f(value)
        }
        return self
    }

Still, this is not a reliable solution: it's not guaranteed, that function f will be invoked before any other mapping function which is invoked due to an update. These kind of issues can be solved only with introducing an "execution context" (for example a dispatch queue) which defines where the mapping function will be executed:

    public func subscribe(ec: ExecutionContext = <default>, f: Result<T> -> Void) -> Signal<T> {
        mutex.lock {
            if let value = self.value {
                ec.execAsync { f(value) }
            }
            callbacks.append(f)
        }
        return self
    }