为客户端设计API到第三方服务

这似乎是Obesrvable模式的一个很好的例子.这种模式来自Reactive Extensions of .NET,但也可用于Java和Scala.该库由Netflix提供,质量非常好.

这种模式具有良好的理论基础 – 它是理论意义上的迭代器的双重性.但更重要的是,它有很多实用的想法.特别是它处理时间非常好,例如你可以限制你想要的事件率.

通过观察,您可以处理非常高级别的事件.在.NET中,它看起来很像SQL查询.您可以注册某些事件(“FROM”),过滤它们(“WHERE”)并最终处理它们(“SELECT”).在Scala中,您可以使用标准的monadic API(map,filter,flatMap),当然也可以使用“for expressions”.

一个例子可以看起来像

stackoverflowQuestions.filter(_.tag == "Scala").map(_.subject).throttleLast(1 second).subscribe(println _)

Obeservable消除了基于事件的系统会遇到的许多问题

>处理订阅
>处理错误
>过滤和预处理事件
>缓冲事件

构建API

您的API应该为您拥有的每个事件源提供一个obesrvable.对于过程调用,您提供了一个将函数调用映射到obesrvable的函数.此函数将调用远程过程并通过obeservable提供结果.

实施细节

将以下依赖项添加到build.sbt：

libraryDependencies +=     "com.netflix.rxjava" % "rxjava-scala" % "0.15.0"

然后,您可以使用以下模式将回调转换为obeservable(假设您的远程API具有某种方式来注册和取消注册回调)：

private val callbackFunc : (rx.lang.scala.Observer[String]) => rx.lang.scala.Subscription = { o =>
  val listener = {
    case Value(s) => o.onNext(s)
    case Error(e) => o.onError(o)
  }

  remote.subscribe(listener)

  // Return an interface to cancel the subscription
  new Subscription {
    val unsubscribed = new AtomicBoolean(false)
    def isUnsubscribed: Boolean = unsubscribed.get()

    val asJavaSubscription: rx.Subscription = new rx.Subscription {
      def unsubscribe() {
        remote.unsubscribe(listener)
        unsubscribed.set(true)
      }
    }
  }

如果您有一些具体问题,请询问,我可以改进答案

额外的资源

Martin Odersky等人有一个非常好的course.在课程中,涵盖了Observables和其他反应技术.