Built-in Pipeline Behaviors

MediatorK ships seven ready-to-use pipeline behaviors. Drop them into MediatorFactory.create, no boilerplate required.

1. LoggingPipelineBehavior: logs every request in and out
2. TimeoutPipelineBehavior: cancels requests that exceed a deadline
3. RequestCounterPipelineBehavior: counts dispatches per request type
4. CachingPipelineBehavior: skips the handler on a cache hit
5. TimingPipelineBehavior: measures handler duration
6. ErrorTrackingPipelineBehavior: forwards exceptions to crash reporters
7. TransactionPipelineBehavior: wraps every request in a begin / commit / rollback transaction

---

LoggingPipelineBehavior

Logs each request as it enters and exits the pipeline, including the result on the exit line. Accepts any (String) -> Unit logger so it works on every platform.

KMP / common (println)

val mediator = MediatorFactory.create(
    registrars = listOf(AppRegistrar()),
    pipelineBehaviors = listOf(
        LoggingPipelineBehavior(logger = ::println),
    ),
)

Output:

→ GetUserQuery
← GetUserQuery result=User(id=user-1, name=Alice)

JVM: SLF4J

import org.slf4j.LoggerFactory

val log = LoggerFactory.getLogger("Mediator")

LoggingPipelineBehavior(logger = log::info)

Android: Logcat

LoggingPipelineBehavior(logger = { msg -> Log.d("Mediator", msg) })

JS / browser: console

LoggingPipelineBehavior(logger = { msg -> console.log(msg) })

Parameters

Parameter	Type	Default	Description
logger	(String) -> Unit	::println	Function that receives each log line
order	Int	-100	Low number = outermost, logs before any other behavior

---

TimeoutPipelineBehavior

Cancels the downstream pipeline if it does not complete within the given deadline. Throws TimeoutCancellationException (a CancellationException) when the deadline is exceeded.

val mediator = MediatorFactory.create(
    registrars = listOf(AppRegistrar()),
    pipelineBehaviors = listOf(
        TimeoutPipelineBehavior(timeoutMillis = 5_000),
    ),
)

Parameters

Parameter	Type	Default	Description
timeoutMillis	Long	required	Maximum allowed duration per dispatch. Must be > 0
order	Int	0	Position in the behavior chain

---

RequestCounterPipelineBehavior

Counts how many times each request type has passed through the pipeline.

val counter = RequestCounterPipelineBehavior()

val mediator = MediatorFactory.create(
    registrars = listOf(AppRegistrar()),
    pipelineBehaviors = listOf(counter),
)

mediator.send(GetUserQuery(id = "user-1"))
mediator.send(GetUserQuery(id = "user-2"))
mediator.send(CreateOrderCommand(cartId = "cart-42"))

counter.countFor(GetUserQuery::class)       // 2
counter.countFor(CreateOrderCommand::class) // 1
counter.snapshot()                          // {"GetUserQuery" to 2, "CreateOrderCommand" to 1}

API

Member	Description
countFor(KClass<*>)	Returns the dispatch count for a specific request class
snapshot()	Returns a copy of all counts as Map<String, Long>
reset()	Clears all counters

Parameters

Parameter	Type	Default	Description
order	Int	0	Position in the behavior chain

---

CachingPipelineBehavior

Caches handler results by request key for a configurable TTL. On a cache hit the handler is skipped entirely. Best suited for query requests whose results change infrequently.

CachingPipelineBehavior(
    ttlMs = 30_000,                      // cache for 30 seconds
    keyFor = { req -> req.toString() },  // default: full toString
)

Use filter to cache most requests while excluding specific types:

CachingPipelineBehavior(
    ttlMs = 600_000,                     // 10 minutes
    filter = { it !is CreateBookingCommand && it !is GetBookingQuery },
)

API	Description
invalidate(key)	Remove a single entry
clear()	Remove all entries
size()	Count of cached entries

Parameters

Parameter	Type	Default	Description
ttlMs	Long	60_000	Time-to-live per entry in milliseconds
keyFor	(Request<*>) -> String	toString()	Cache key function
filter	(Request<*>) -> Boolean	{ true }	When false, the request bypasses caching entirely
order	Int	0	Position in the behavior chain

---

TimingPipelineBehavior

Measures how long each request takes and reports it via a callback. Timing is always reported, even when the handler throws.

// KMP / println
TimingPipelineBehavior(onTiming = { name, ms -> println("$name took ${ms}ms") })

// Android — Firebase Performance
TimingPipelineBehavior(onTiming = { name, ms ->
    FirebasePerformance.getInstance().newTrace(name).also { it.start(); it.stop() }
})

// JVM — Micrometer
TimingPipelineBehavior(onTiming = { name, ms ->
    meterRegistry.timer(name).record(ms, TimeUnit.MILLISECONDS)
})

Parameters

Parameter	Type	Default	Description
onTiming	(String, Long) -> Unit	required	Callback with (requestName, durationMs)
order	Int	0	Position in the behavior chain

---

ErrorTrackingPipelineBehavior

Intercepts every unhandled exception, forwards it to a callback, then rethrows. Use this to wire crash-reporting services without touching handler code.

// Android — Firebase Crashlytics
ErrorTrackingPipelineBehavior { request, error ->
    FirebaseCrashlytics.getInstance().recordException(error)
}

// KMP — Sentry
ErrorTrackingPipelineBehavior { request, error ->
    Sentry.captureException(error)
}

Parameters

Parameter	Type	Default	Description
order	Int	Int.MAX_VALUE	Innermost by default, fires closest to the handler
onError	(Request<*>, Throwable) -> Unit	required	Callback with the request and the exception

---

TransactionPipelineBehavior

Wraps every request in a transaction: begin() before the handler runs, commit() on success, and rollback() on any exception (which is then rethrown). Implement the small TransactionProvider interface for whatever resource supports begin / commit / rollback — a SQL database, an in-memory store, or your own unit-of-work.

// Adapt your database to the TransactionProvider contract
class DatabaseTransactionProvider(private val db: Database) : TransactionProvider {
    override suspend fun begin() = db.beginTransaction()
    override suspend fun commit() = db.commitTransaction()
    override suspend fun rollback() = db.rollbackTransaction()
}

val mediator = MediatorFactory.create(
    registrars = listOf(AppRegistrar()),
    pipelineBehaviors = listOf(
        TransactionPipelineBehavior(DatabaseTransactionProvider(db)),
    ),
)

Parameters

Parameter	Type	Default	Description
transactionProvider	TransactionProvider	required	Resource that supports begin(), commit(), and rollback()

---

Registering multiple built-in behaviors

val counter = RequestCounterPipelineBehavior()

val mediator = MediatorFactory.create(
    registrars = listOf(AppRegistrar()),
    pipelineBehaviors = listOf(
        LoggingPipelineBehavior(logger = ::println, order = -100),
        TimeoutPipelineBehavior(timeoutMillis = 5_000, order = -1),
        counter,
    ),
)

---

Next

→ Pre / Post Behaviors