What is the equivalent for PublishSubject in ReactiveSwift?

Question

I am using ReactiveSwift in my project, and I'm wondering what is the equivalent for PublishSubject?

for example in RXSwift we can do:

let disposeBag = DisposeBag()

let pubSubj = PublishSubject<String>()

pubSubj.on(.next("(next 1")) //event emitted to no subscribers

pubSubj.subscribe({ //subscriber added, but no replay of "next 1"
                print("line: \(#line),", "event: \($0)")
            })
.disposed(by: disposeBag)

pubSubj.on(.next("(next 2")) //event emitted and received by subscriber
pubSubj.onError(MyError.error1) //emits error and terminates sequence

pubSubj.on(.next("next 3")) //pubSubj cannot emit this event
/* prints: 
line: 8, event: next((next 2)
line: 8, event: error(error1) 
*/

Answer 1

ReactiveSwift doesn't have a single type like Subject that can be both the input and output for a signal. Instead you can use Signal.pipe() to create an input and its corresponding output signal:

import ReactiveSwift

enum MyError: Error { case error1 }

let (lifetime, token) = Lifetime.make()

let (signal, input) = Signal<String, MyError>.pipe()

input.send(value: "1")

signal
    .take(during: lifetime)
    .observe {
        print("line: \(#line),", "event: \($0)")
    }

input.send(value: "2")
input.send(error: .error1)

input.send(value: "3")

/* prints:
line: 14, event: VALUE 2
line: 14, event: FAILED error1
*/