Couple updates.

build.sbt

@@ -6,7 +6,7 @@ android.Plugin.androidBuildAar
 
 name := "geteit-app"
 organization := "com.geteit"
-version := "0.2"
+version := "0.3-SNAPSHOT"
 
 scalaVersion := "2.11.7"
 
@@ -37,7 +37,7 @@ libraryDependencies ++= Seq (
   "com.android.support" % "cardview-v7" % supportLibVersion,
   "com.android.support" % "recyclerview-v7" % supportLibVersion,
   "com.nineoldandroids" % "library" % "2.4.0",
-  "com.geteit" %% "geteit-utils" % "0.4-SNAPSHOT",
+  "com.geteit" %% "geteit-utils" % "0.5-SNAPSHOT",
   "com.koushikdutta.async" % "androidasync" % "2.1.3",
   "org.scalatest" %% "scalatest" % "2.2.5" % Test,
   "org.scalacheck" %% "scalacheck" % "1.11.6" % Test,

project/plugins.sbt

@@ -2,4 +2,4 @@ addSbtPlugin("org.xerial.sbt" % "sbt-sonatype" % "0.5.0")
 
 addSbtPlugin("com.jsuereth" % "sbt-pgp" % "1.0.0")
 
-addSbtPlugin("com.hanhuy.sbt" % "android-sdk-plugin" % "1.5.14")
+addSbtPlugin("com.hanhuy.sbt" % "android-sdk-plugin" % "1.5.16")

src/main/scala/com/geteit/cache/CacheStorage.scala

@@ -6,11 +6,11 @@ import java.lang.System._
 import android.content.Context
 import com.geteit.cache.CacheEntryData.CacheEntryDao
 import com.geteit.concurrent.Threading
-import com.geteit.db.{CachedStorage, Storage}
+import com.geteit.db.CachedStorage
 import com.geteit.events.EventContext
+import com.geteit.inject.{Injectable, Injector}
 import com.geteit.util.Log._
 import com.geteit.util._
-import com.geteit.inject.{Injectable, Injector}
 
 import scala.concurrent.Future
 
@@ -38,7 +38,7 @@ class CacheStorage(implicit inj: Injector) extends CachedStorage[String, CacheEn
 
   override def get(key: String): Future[Option[CacheEntryData]] = super.get(key) map filterValid
 
-  override def getAll(keys: Seq[String]): Future[Seq[Option[CacheEntryData]]] = super.getAll(keys) map { _ map filterValid }
+  override def getAll(keys: Traversable[String]): Future[Seq[Option[CacheEntryData]]] = super.getAll(keys) map { _ map filterValid }
 
   def entryFile(id: Uid) = CacheStorage.entryFile(cacheDir, id)
 

src/main/scala/com/geteit/content/KeyValueStorage.scala

@@ -27,7 +27,10 @@ class KeyValueStorage(implicit inj: Injector) extends CachedStorage[String, KeyV
       case Some(_) => Future.successful(())
     } .recoverWithLog()
 
-    val source = signal(key)(ev).map { kv => LoggedTry(kv.decode[T]).getOrElse(default) }
+    val source = signal(key)(ev).map {
+      case Some(kv) => LoggedTry(kv.decode[T]).getOrElse(default)
+      case None => default
+    }
 
     new ProxySignal[T](source) with Source[T] {
       override protected def computeValue(current: Option[T]) = source.currentValue

src/main/scala/com/geteit/db/CachedStorage.scala

@@ -67,7 +67,7 @@ abstract class CachedStorage[K, V](implicit val dao: Dao[K, V], inj: Injector) e
 
   def listAll = storage.read { dao.list(_) }
 
-  def getAll(keys: Seq[K]): Future[Seq[Option[V]]] = {
+  def getAll(keys: Traversable[K]): Future[Seq[Option[V]]] = {
     val cachedEntries = keys.flatMap { key => Option(cache.get(key)) map { value => (key, value) } }.toMap
     val missingKeys = keys.toSet -- cachedEntries.keys
     val loaderOfMissing = { (keys: Seq[K], db: SQLiteDatabase) => keys.flatMap(key => dao.get(key)(db).map { value => (key, value) })}
@@ -80,9 +80,9 @@ abstract class CachedStorage[K, V](implicit val dao: Dao[K, V], inj: Injector) e
         }
       } .toMap
 
-      keys map { key =>
+      keys .map { key =>
         loadedMap.get(key).orElse(cachedEntries.get(key)).getOrElse(None)
-      }
+      } (breakOut)
     }
   }
 
@@ -96,18 +96,17 @@ abstract class CachedStorage[K, V](implicit val dao: Dao[K, V], inj: Injector) e
     prev.fold { addInternal(key, creator) } { v => updateInternal(key, updater)(v).map(_.fold(v)(_._2)) }
   }
 
-  def updateAll(updaters: scala.collection.Map[K, V => V]): Future[Seq[(V, V)]] = {
-    val keys = updaters.keys.toSeq
+  def updateAll(keys: Iterable[K], updater: V => V): Future[Seq[(V, V)]] = {
     getAll(keys) flatMap { values =>
-      val updated = keys.zip(values) flatMap { case (k, v) =>
+      val updated = keys.zip(values) .flatMap { case (k, v) =>
         Option(cache.get(k)).flatten.orElse(v).flatMap { value =>
-          val updated = updaters(k)(value)
+          val updated = updater(value)
           if (updated != value) {
             cache.put(k, Some(updated))
             Some(value -> updated)
           } else None
         }
-      }
+      } .toSeq
 
       if (updated.isEmpty) Future.successful(Seq.empty)
       else
@@ -117,17 +116,16 @@ abstract class CachedStorage[K, V](implicit val dao: Dao[K, V], inj: Injector) e
     }
   }
 
-  def updateOrCreateAll(updaters: K Map (Option[V] => V)): Future[Set[V]] = {
-    val keys = updaters.keys.toSeq
+  def updateOrCreateAll(keys: Iterable[K], updater: ((K, Option[V]) => V)): Future[Set[V]] = {
     getAll(keys) flatMap { values =>
       val loaded: Map[K, Option[V]] = keys.zip(values).map { case (k, v) => k -> Option(cache.get(k)).flatten.orElse(v) } (breakOut)
       val toSave = Seq.newBuilder[V]
       val added = Seq.newBuilder[V]
       val updated = Seq.newBuilder[(V, V)]
 
-      val result = updaters .map { case (key, updater) =>
+      val result = keys .map { key =>
         val current = loaded.get(key).flatten
-        val next = updater(current)
+        val next = updater(key, current)
         current match {
           case Some(c) if c != next =>
             cache.put(key, Some(next))
@@ -193,6 +191,13 @@ trait CachedStorageSignal[K, V] { self: CachedStorage[K, V] =>
 
   private def loadAll: Future[Map[K, V]] = listAll map { _.map(v => dao.getId(v) -> v)(breakOut) }
 
+  def onChange(key: K) = {
+    val added = onAdded.map[Seq[Cmd[K, V]]](_.filter(v => dao.getId(v) == key).map(Add(_))).filter(_.nonEmpty)
+    val updated = onUpdated.map[Seq[Cmd[K, V]]](_.filter(v => dao.getId(v._2) == key).map(p => Add(p._2))).filter(_.nonEmpty)
+    val removed = onRemoved.map[Seq[Cmd[K, V]]](_.filter(_ == key).map(Del(_))).filter(_.nonEmpty)
+    EventStream.union[Seq[Cmd[K, V]]](added, updated, removed)
+  }
+
   def all(implicit ev: EventContext): Signal[Map[K, V]] = new AggregatingSignal[Seq[Cmd[K, V]], Map[K, V]](onChanged, loadAll, { (values, cmds) =>
     var res = values
     cmds foreach {
@@ -202,25 +207,12 @@ trait CachedStorageSignal[K, V] { self: CachedStorage[K, V] =>
     res
   })
 
-  def signal(id: K)(implicit ev: EventContext): Signal[V] = new Signal[V]() {
-
-    def reload() = self.get(id).map { _.foreach(this.publish) }
+  def signal(id: K)(implicit ev: EventContext): Signal[Option[V]] = {
+    val added = onAdded.map(_.reverseIterator.find(v => dao.getId(v) == id)).filter(_.nonEmpty)
+    val updated = onUpdated.map(_.reverseIterator.collectFirst { case (_, v) if dao.getId(v) == id => v }).filter(_.nonEmpty)
+    val removed = onRemoved.filter(_.contains(id)).map(_ => Option.empty[V])
 
-    private var observers = Seq.empty[Subscription]
-
-    override protected def onWire(): Unit = {
-      observers = Seq(
-        onAdded.map(_.reverseIterator.find(v => dao.getId(v) == id)) { _.foreach(publish) }, // FIXME: possible race condition with reload result
-        onUpdated.map(_.reverseIterator.find(p => dao.getId(p._2) == id)) { _.foreach(p => publish(p._2)) },
-        onRemoved.filter(_.exists(_ == id)) { _ => reload() }
-      )
-      reload()
-    }
-
-    override protected def onUnwire(): Unit = {
-      observers foreach (_.destroy())
-      clear()
-    }
+    Signal.wrap(EventStream.union(added, updated, removed)).orElse(Signal.future(get(id)))
   }
 
   def findSignal(matcher: Matcher[V])(implicit ev: EventContext): Signal[Set[K]] = new Signal[Set[K]] {
@@ -272,8 +264,8 @@ trait CachedStorageSignal[K, V] { self: CachedStorage[K, V] =>
 
 object CachedStorageSignal {
   trait Cmd[+K, +V]
-  case class Add[A](v: A) extends Cmd[Nothing, A]
-  case class Del[A](v: A) extends Cmd[A, Nothing]
+  case class Add[V](v: V) extends Cmd[Nothing, V]
+  case class Del[K](k: K) extends Cmd[K, Nothing]
 }
 
 class AutoCloseCursor(c: Cursor) extends CursorWrapper(c) {

src/main/scala/com/geteit/inject/GtAppModule.scala

@@ -14,6 +14,7 @@ object GtAppModule {
     bind [ImageProvider] to new BasicImageProvider
     bind [CacheService] to new CacheService
     bind [CacheStorage] to new CacheStorage
+    bind [UserAgent] to UserAgent(inject[Context])
     bind [AsyncClient] to new AsyncClient
     bind [CookieStorage] to new MemoryCookieStorage
     bind [KeyValueStorage] to new KeyValueStorage

src/main/scala/com/geteit/net/AsyncClient.scala

@@ -29,7 +29,7 @@ class AsyncClient(implicit inj: Injector) extends Injectable {
 
   lazy val bodyDecoder = inject[ResponseBodyDecoder]
 
-  lazy val userAgent = AsyncClient.userAgent(inject[Context])
+  lazy val userAgent = inject[UserAgent]
 
   val client = ClientWrapper { new AsyncHttpClient(new AsyncServer) }
 
@@ -75,7 +75,7 @@ class AsyncClient(implicit inj: Injector) extends Injectable {
     val r = new AsyncHttpRequest(req.uri, req.httpMethod)
     r.setTimeout(req.timeout.toMillis.toInt)
     req.headers.foreach(p => r.addHeader(p._1, p._2))
-    r.setHeader("User-Agent", userAgent)
+    r.setHeader("User-Agent", req.headers.getOrElse("User-Agent", userAgent.str))
     req.getBody(r)
   }
 
@@ -89,22 +89,18 @@ class AsyncClient(implicit inj: Injector) extends Injectable {
       case _ => ContentRange(0, contentLength, contentLength)
     }
 
-    debug(s"got connection response for request: $uri, status: '$httpStatus', length: '$contentLength', headers: '${response.headers()}'")
-    verbose(s"decoder: $decoder")
+    verbose(s"got connection response for request: $uri, status: '$httpStatus', length: '$contentLength', headers: '${response.headers()}'")
 
     progressCallback foreach (_(Progress(0L, range, Progress.Running)))
     if (contentLength == 0) CancellableFuture.successful(Response(httpStatus))
     else {
-      debug(s"waiting for content from $uri")
-
       val p = Promise[Response]()
       val consumer = decoder(response.headers(), contentLength)
 
       response.setDataCallback(new DataCallback {
         var bytesReceived = new AtomicLong(0L)
 
         override def onDataAvailable(emitter: DataEmitter, bb: ByteBufferList): Unit = {
-          debug(s"data received for $uri, length: ${bb.remaining}")
           val numConsumed = bb.remaining
           consumer.consume(bb)
           progressCallback foreach { cb => Future(cb(Progress(bytesReceived.addAndGet(numConsumed), range, Progress.Running))) }
@@ -113,7 +109,7 @@ class AsyncClient(implicit inj: Injector) extends Injectable {
 
       response.setEndCallback(new CompletedCallback {
         override def onCompleted(ex: Exception): Unit = {
-          debug(s"response for $uri ENDED, ex: $ex, p.isCompleted: ${p.isCompleted}")
+          verbose(s"onCompleted(ex: $ex) $uri ")
           if (ex != null) ex.printStackTrace(Console.err)
           response.setDataCallback(null)
           p.tryComplete(
@@ -150,16 +146,20 @@ object AsyncClient {
   val DefaultTimeout = 5.minutes
   val EmptyHeaders = Map[String, String]()
 
-  def userAgent(implicit context: Context) = {
-    import android.os.Build._
-    val appVersion = context.getPackageManager.getPackageInfo(context.getPackageName, 0).versionName
-    s"${context.getPackageName}/$appVersion (Android ${VERSION.RELEASE}; $MANUFACTURER $MODEL)"
-  }
-
   private def exceptionStatus: PartialFunction[Throwable, Response] = {
     case CancelException => throw CancelException
     case e: ConnectException => Response(Response.ConnectionError(e.getMessage))
     // TODO: handle connection exceptions
     case NonFatal(e) => Response(Response.InternalError(e.getMessage, Some(e)))
   }
 }
+
+case class UserAgent(str: String)
+
+object UserAgent {
+  def apply(context: Context): UserAgent = {
+    import android.os.Build._
+    val appVersion = context.getPackageManager.getPackageInfo(context.getPackageName, 0).versionName
+    UserAgent(s"${context.getPackageName}/$appVersion (Android ${VERSION.RELEASE}; $MANUFACTURER $MODEL)")
+  }
+}

src/main/scala/com/geteit/view/RoboView.scala

@@ -12,7 +12,7 @@ import scala.collection.mutable
 trait RoboView { self: TextView =>
   import RoboView._
 
-  protected def initFont(attrs: AttributeSet)(implicit context: Context) {
+  protected def initFont(attrs: AttributeSet)(implicit context: Context) = {
     val a = context.obtainStyledAttributes(attrs, R.styleable.RoboView)
     val typeId = a.getInt(R.styleable.RoboView_fontFamily, 0)
     setTypeface(roboTypeface(typeId))