-
Notifications
You must be signed in to change notification settings - Fork 5
/
MDCPropagatingDispatcherConfigurator.scala
90 lines (76 loc) · 2.69 KB
/
MDCPropagatingDispatcherConfigurator.scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package com.github.rishabh9
import java.util.concurrent.TimeUnit
import akka.dispatch._
import com.typesafe.config.Config
import org.slf4j.MDC
import scala.concurrent.ExecutionContext
import scala.concurrent.duration.{Duration, FiniteDuration}
/**
* Configurator for a MDC propagating dispatcher.
*
* To use it, configure play like this:
* {{{
* play {
* akka {
* actor {
* default-dispatcher = {
* type = "com.github.rishabh9.MDCPropagatingDispatcherConfigurator"
* }
* }
* }
* }
* }}}
*
* Credits to James Roper for the [[https://github.com/jroper/thread-local-context-propagation/ initial implementation]]
*/
class MDCPropagatingDispatcherConfigurator(config: Config, prerequisites: DispatcherPrerequisites)
extends MessageDispatcherConfigurator(config, prerequisites) {
private val instance = new MDCPropagatingDispatcher(
this,
config.getString("id"),
config.getInt("throughput"),
FiniteDuration(config.getDuration("throughput-deadline-time", TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS),
configureExecutor(),
FiniteDuration(config.getDuration("shutdown-timeout", TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS))
override def dispatcher(): MessageDispatcher = instance
}
/**
* A MDC propagating dispatcher.
*
* This dispatcher propagates the MDC current request context if it's set when it's executed.
*/
class MDCPropagatingDispatcher(_configurator: MessageDispatcherConfigurator,
id: String,
throughput: Int,
throughputDeadlineTime: Duration,
executorServiceFactoryProvider: ExecutorServiceFactoryProvider,
shutdownTimeout: FiniteDuration)
extends Dispatcher(_configurator, id, throughput, throughputDeadlineTime, executorServiceFactoryProvider, shutdownTimeout) {
self =>
override def prepare(): ExecutionContext = new ExecutionContext {
// capture the MDC
val mdcContext = MDC.getCopyOfContextMap
def execute(r: Runnable) = self.execute(new Runnable {
def run() = {
// backup the callee MDC context
val oldMDCContext = MDC.getCopyOfContextMap
// Run the runnable with the captured context
setContextMap(mdcContext)
try {
r.run()
} finally {
// restore the callee MDC context
setContextMap(oldMDCContext)
}
}
})
def reportFailure(t: Throwable) = self.reportFailure(t)
}
private[this] def setContextMap(context: java.util.Map[String, String]) {
if (context == null) {
MDC.clear()
} else {
MDC.setContextMap(context)
}
}
}