我有一个基本的scala akka http CRUD应用程序 . 请参阅下面的相关课程 .
我只想将实体id和一些数据(作为json)写入Kafka主题,例如,创建/更新实体 .
我正在看http://doc.akka.io/docs/akka-stream-kafka/current/producer.html,但我是scala和akka的新手,并且不确定如何将它集成到我的应用程序中?
例如,从上面的文档中,这是一个写入kafka的 生产环境 者的例子,所以我认为我需要类似的东西,但是我的应用程序中的行踪应该如何?我可以在创建用户后在我的服务中的create方法中添加另一个 Map 调用吗?
非常感谢!
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)
}
Router.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
}
Service.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
}
Repo.scala
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())))
)
}
1 回答
既然你已经从
Entity
编写了Route
来解组1User
,我认为你不需要Producer.plainSink
. 相反,我认为Producer.send也会起作用 . 另外,作为附注,抛出异常不是"idiomatic" scala . 所以我改变了无效用户的逻辑: