0

私は基本的な 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)

}

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())))
    )
  }
4

1 に答える 1

2

からRoute1 つだけアンマーシャリングするように書いたので、必要ないと思います。むしろ、うまくいくと思います。また、補足として、例外のスローは「慣用的な」scala ではありません。そこで、無効なユーザーのロジックを変更しました。UserEntityProducer.plainSinkProducer.send

val producer : KafkaProducer = new KafkaProducer(producerSettings)

val routes : Route = 
  post {
    entity(as[User]) { user =>
      val createUser = userService.create(user)
      onSuccess(createUser) {
        case Invalid(y: NonEmptyList[Err]) =>  
          complete(BadRequest -> "invalid user")
        case Valid(u: User) => { 
          val producerRecord = 
            new ProducerRecord[Array[Byte], String]("topic1",s"""{"userId" : ${u.userId}, "entity" : "User"}""")

          onComplete(producer send producerRecord) { _ =>
            complete(ToResponseMarshallable((StatusCodes.Created, u)))
          }
        }
      }
    }
  }
于 2017-09-07T15:32:44.813 に答える