Composition of Observables in rxhs

LongPolling.hs

getLastPublishedMessages :: PubSub -> ClientId -> IO (Either SomeException [WebResponse])
getLastPublishedMessages pubsub cid = do
    msubject <- getClientSubject pubsub cid
    case msubject of
      Just subject ->
        Observable.toEither .
          Observable.first .
          Observable.filter (not . null) .
          Observable.bufferInterval (Time.microSeconds 1000) .
          Observable.concatMap excludeClientId $
          Observable.toObservable subject
      Nothing ->
        return
          $ Left
          $ SomeException
          $ ErrorCall "unknown clientId"
  where
    excludeClientId req =
      if _requestClientId req == cid then [] else [Request $ fmap Aeson.toJSON req]