vhutov

trait DataStore[F[_]] {
  def get(id: String)(implicit ctx: LogContext)
  def update[T](id: String, t: T)(implicit ctx: LogContext)
  def delete(id: String)(implicit ctx: LogContext)
}
class MongoDataStore[F[_]: Async] extends DataStore[F] {
  override def get(id: String)(implicit ctx: LogContext)
  override def update[T](id: String, t: T)(implicit ctx: LogContext)
  override def delete(id: String)(implicit ctx: LogContext)
}

vhutov

class LogContext {
  val mdc: Map[String, String] = Map()
  def update(k: String, v: String): LogContext = ???
}

vhutov

def stage1(i: Int): IO[Int] = a()
def stage2(i: Int): IO[Int] = b()
def stage3(i: Int): IO[Int] = c().withContext(_.name("op"))

stage1(0)
  .withContext(_.id(0))        // MDC {id: 0}
  .flatMap(i => 
    stage2(i)
      .withContext(_.value(i)) // MDC {id: 0, value: i}
  )

vhutov

import scala.concurrent.duration._
import io.gatling.core.Predef._
import Predef._

class Test extends Simulation {
  
  
  val kafkaP = kafkaProtocol.properties(Map("bootstrap.servers" -> "localhost:9092"))
  
  val jmsP = jmsProtocol.url("localhost:7222")

vhutov

object KafkaProtocolBuilder {

  implicit def toProtocol(builder: KafkaProtocolBuilder): KafkaProtocol = builder.build

  def apply(configuration: GatlingConfiguration): KafkaProtocolBuilder =
    KafkaProtocolBuilder(KafkaProtocol(configuration))
}

case class KafkaProtocolBuilder(kafkaProtocol: KafkaProtocol) {
  def build = kafkaProtocol

vhutov

class JmsManagerActor(messageParser: MessageParser) extends Actor {

  def receive: Receive = {
    case SubscribeForMessage(id) if isAlreadyReceived(id) =>
      msgsBuffer.remove(id)
      sender ! Ack

    case SubscribeForMessage(id) =>
      subscriptionMap.update(id, sender())

vhutov

class GenericAction[V <: Message](
  val coreComponents: CoreComponents,
  val attributes: Attributes[V],
  val publisher: Publisher[V],
  val consumer: Consumer,
  val throttled: Boolean,
  val next: Action
) extends ExitableAction with NameGen {

  val statsEngine = coreComponents.statsEngine

vhutov

class GenericActionBuilder[V <: Message](
  attributes: Attributes[V],
  requestActionBuilder: RequestActionBuilder[V],
  responseActionBuilder: ResponseActionBuilder
) extends ActionBuilder with LazyLogging {

  override def build(ctx: ScenarioContext, next: Action): Action = {
    import ctx._

    val requestProtocolComponents = protocolComponentsRegistry.components(requestActionBuilder.key)

vhutov

class KafkaRequestActionBuilder[K, V <: Message](
  topic: String,  //this parameters will be provided when building KafkaPublisher
  keyExtractor: V => Option[K]) extends RequestActionBuilder[V] {
  
  override type PK = KafkaProtocol.KafkaProtocolKey.type
  override def key = KafkaProtocol.KafkaProtocolKey

  override def build(ctx: ScenarioContext, components: KafkaComponents): Producer[V] = {
    val producer = new KafkaProducer[K, V](components.protocol.properties.asJava)
    new KafkaPublisher[K, V](producer, topic)(keyExtractor)

vhutov

trait RequestActionBuilder[V <: Message] {
  type PK <: ProtocolKey
  def key: PK

  def build(ctx: ScenarioContext, components: PK#Components): Producer[V]
}