save service id

src/main/scala/com.gmei.up/Main.scala

@@ -30,7 +30,7 @@ class UserInfoDeserializationSchema extends KafkaDeserializationSchema[UserInfo]
 }
 
 // TODO remove
-// extends LazyLogging 
+// extends LazyLogging
 object Main {
   def main(args: Array[String]): Unit =
     // println("###############")
@@ -109,7 +109,7 @@ object Main {
           Redis.save(ES.request(diaryReq), deviceId, "diary", streamTimeBegin)
           Redis.save(ES.request(tractateReq), deviceId, "tractate", streamTimeBegin)
           Redis.save(ES.request(answerReq), deviceId, "answer", streamTimeBegin)
-          Redis.save(ES.request(serviceDiaryReq), deviceId, "service_diary", streamTimeBegin)
+          Redis.saveServiceDiary(ES.request(serviceDiaryReq), deviceId)
 
           // current time
           // TODO logger

src/main/scala/com.gmei.up/utils/ES.scala

@@ -57,7 +57,6 @@ object ES {
   implicit object ContentHitReader extends HitReader[Content] {
     override def read(hit: Hit): Try[Content] = {
       val source = hit.sourceAsMap
-      println(source)
       val serviceId = source.getOrElse("service", Map[String, Long]()).asInstanceOf[Map[String, Long]].get("id")
       Try(
         Content(source("id").toString.toLong, hit.index, serviceId, source("tags_v3").toString)
@@ -69,13 +68,12 @@ object ES {
 
   def request(req: SearchRequest): Either[Throwable, Future[IndexedSeq[Content]]] = {
     val client: ElasticClient = ESClient.create("172.16.52.33", 9200, "elastic", "gengmei!@#")
-    println("@@@@@@@@@@@@@")
-    client.execute(req).map { resp =>
-      val a = resp.result.to[Content]
-      a.foreach(x => println("################", x.id, x.index, x.serviceId))
-    }
-    println("@@@@@@@@@@@@@")
-
+    // println("@@@@@@@@@@@@@")
+    // client.execute(req).map { resp =>
+    //   val a = resp.result.to[Content]
+    //   a.foreach(x => println("################", x.id, x.index, x.serviceId))
+    // }
+    // println("@@@@@@@@@@@@@")
     try {
       Right(client.execute(req).map(resp => resp.result.to[Content]))
     } catch {

src/main/scala/com.gmei.up/utils/Redis.scala

@@ -60,4 +60,26 @@ object Redis {
         }
     }
   }
+
+  def saveServiceDiary(
+      contentEitherFuture: Either[Throwable, Future[IndexedSeq[Content]]],
+      deviceId: String
+  ): Unit = {
+    val key = s"streaming:candidate:service_diary:device_id:${deviceId}"
+    contentEitherFuture match {
+      case Left(e) => e.printStackTrace()
+      case Right(contentFuture) =>
+        pRc4.withClient { client =>
+          contentFuture.foreach { seq =>
+            val idPaids = seq.map(c => (c.id, c.serviceId.getOrElse(-1L)))
+            if (idPaids.size > 0) {
+              client.del(key)
+              idPaids.foreach { pair =>
+                client.rpush(key, pair)
+              }
+            }
+          }
+        }
+    }
+  }
 }