title: Creating a user-controllable RxSwift Observer
body: <p>I'm trying to implement user-driven refreshing in my Rx based networking code, and my current design is as follows:</p>
  
  <ol>
  <li>Create a sink that has <code>Void</code> values passed into it every time the user initiates a refresh action</li>
  <li><code>flatMap</code> the latest <code>.Next</code> event on that sink's <code>Observable</code> into a new network call</li>
  <li>Transform the network response into a new view model and pass that back into the view controller</li>
  </ol>
  
  <p>The part I'm getting hung up on is how to create a sink for those events to go down. My current code is as follows:</p>
  
  <pre><code>func contactListModel() -&gt; Observable&lt;ContactListViewModel&lt;Contact&gt;&gt; {
      // Create a sink for refresh events
      var refreshSink: AnyObserver&lt;Void&gt; = AnyObserver { event in }
      let refreshObservable = Observable&lt;Void&gt;.create { observer in
          refreshSink = observer
          return NopDisposable.instance
      }
  
      // Define action handlers
      let searchClosure = { (query: String?) in
          self.contactsSearchTerm.value = query
      }
      let refreshClosure = refreshSink.onNext
  
      // TODO: [RP] Make contact list view controller handle a nil view model to remove the need for this code
      let initialViewModel = ContactListViewModel&lt;Contact&gt;(contacts: [], searchClosure: searchClosure, refreshClosure: refreshClosure)
  
      // Perform an initial refresh
      defer {
          refreshSink.onNext()
      }
  
      // Set up subscription to push a new view model each refresh
      return refreshObservable
          .flatMapLatest {
              return self.networking.request(.ListContacts)
          }
          .mapToObject(ListContactsResponse)
          .map { response in
              return ContactListViewModel(contacts: response.contacts, searchClosure: searchClosure, refreshClosure: refreshClosure)
          }
          .startWith(initialViewModel)
  }
  </code></pre>
  
  <p>Now it's obvious why my code to create an event sink doesn't work here. The block being passed into <code>refreshObservable</code>'s <code>create</code> method is only called once the observer is subscribed to, so the <code>refreshSink</code> won't be reassigned until then. Furthermore, if this observable is subscribed to more than once, the <code>refreshSink</code> variable will be reassigned.</p>
  
  <p>So my question is this: <em>how do I create an <code>Observable</code> that I can manually push events down? Or alternatively, is there a better design I could be using here?</em></p>
  
  <p>I know ReactiveCocoa has the <code>pipe</code> static method on <code>Signal</code> that will do something like what I'm looking for, but I've found no equivalent in the Rx API.</p>