Add registration cleaning worker #239
Conversation
| .map(_.getBody) | ||
| .map(Json.parse) | ||
| .flatMap(_.validate[InvalidTokens].asOpt) | ||
| .foldLeft(List.empty[String]){ case (agg, value) => agg ++ value.tokens } |
There was a problem hiding this comment.
why not .flatMap(_.tokens) rather than foldLeft
There was a problem hiding this comment.
Now that you mention it, yeah why? 😄
| val processReport = traverseAndDelete(tokens, CleaningResult(0, 0, 0)).unsafeRunSync() | ||
|
|
||
| logger.info(s"Deleted ${processReport.deletedRows} rows, deleted ${processReport.deletedRegistrations} registration and failed ${processReport.failures} time(s)") | ||
| } |
There was a problem hiding this comment.
I will keep using streams to handle the logic. It make it quite easy to follow the logic flow:
def printResult(result: CleaningResult): IO[Unit] = IO(
logger.info(s"Deleted ${result.deletedRows} rows, deleted ${result.deletedRegistrations} registration and failed ${result.failures} time(s)")
)
val tokens: List[String] = input
.getRecords.asScala
.map(_.getBody)
.map(Json.parse)
.flatMap(_.validate[InvalidTokens].asOpt)
.flatMap(_.tokens)
.toList
Stream
.emits(tokens).covary[IO]
.evalMap(deleteAndSwallowError) // or .parEvalMapUnordered(awsApiConcurrencyLimit)(deleteAndSwallowError) for parallel execution
.reduce(_ combined _)
.evalTap(printResult)
.compile
.drain
.unsafeRunSync()
There was a problem hiding this comment.
Got it, Streams still look like alien technology to me but I'll get used to it.
|
|
||
| object InvalidTokens { | ||
| implicit val invalidTokensJF: Format[InvalidTokens] = Json.format[InvalidTokens] | ||
| } |
There was a problem hiding this comment.
APNS provides a token invalidation timestamp that could be use to avoid discarding a token that would have been refreshed since last time it was fetched. I am not sure we need to handle this edge case though
There was a problem hiding this comment.
Yes I'm hesitating to implement that, I'm worried it might introduce more edge cases
Receives events via SQS.
The events aren't sent yet, I'll add that in a subsequent PR