我有一个基本的 scala akka http CRUD 应用程序。请参阅下面的相关课程。
例如,每当创建/更新实体时,我只想将实体 ID 和一些数据(作为 json)写入 Kafka 主题。
我正在查看http://doc.akka.io/docs/akka-stream-kafka/current/producer.html,但对 scala 和 akka 是新手,不确定如何将其集成到我的应用程序中?
例如,从上面的文档中,这是生产者写信给 kafka 的示例,所以我认为我需要类似的东西,但是我的应用程序中的下落应该去哪里?创建用户后,我可以在我的服务的 create 方法中添加另一个地图调用吗?
非常感谢!
val done = Source(1 to 100)
.map(_.toString)
.map { elem =>
new ProducerRecord[Array[Byte], String]("topic1", elem)
}
.runWith(Producer.plainSink(producerSettings))
或者我是否需要在我的 Server.scala 的 bindAndHandle() 方法中执行类似https://github.com/hseeberger/accessus的示例?
WebServer.scala
object System {
implicit val system = ActorSystem()
implicit val dispatcher = system.dispatcher
implicit val actorMaterializer = ActorMaterializer()
}
object WebServer extends App {
import System._
val config = new ApplicationConfig() with ConfigLoader
ConfigurationFactory.setConfigurationFactory(new LoggingConfFileConfigurationFactory(config.loggingConfig))
val injector = Guice.createInjector(new MyAppModule(config))
val routes = injector.getInstance(classOf[Router]).routes
Http().bindAndHandle(routes, config.httpConfig.interface, config.httpConfig.port)
}
路由器.scala
def routes(): Route = {
post {
entity(as[User]) { user =>
val createUser = userService.create(user)
onSuccess(createUser) {
case Invalid(y: NonEmptyList[Err]) => {
throw new ValidationException(y)
}
case Valid(u: User) => {
complete(ToResponseMarshallable((StatusCodes.Created, u)))
}
}
}
} ~
// More routes here, left out for example
}
服务.scala
def create(user: User): Future[MaybeValid[User]] = {
for {
validating <- userValidation.validateCreate(user)
result <- validating match {
case Valid(x: User) =>
userRepo.create(x)
.map(dbUser => Valid(UserConverters.fromUserRow(x)))
case y: DefInvalid =>
Future{y}
}
} yield result
}
回购协议
def create(user: User): Future[User] = {
mutateDbProvider.db.run(
userTable returning userTable.map(_.userId)
into ((user, id) => user.copy(userId = id)) +=
user.copy(createdDate = Some(Timestamp.valueOf(LocalDateTime.now())))
)
}