Skip to content

Commit 0ef0d75

Browse files
pme123pme123
committed
Worker C8 state of work.
1 parent 751432c commit 0ef0d75

File tree

12 files changed

+202
-56
lines changed

12 files changed

+202
-56
lines changed

‎03-api/src/main/scala/camundala/api/ApiProjectConfig.scala

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,8 @@ case class VersionConfig(major: Int, minor: Int, patch: Int, isSnapshot: Boolean
7878
major * 100000 + minor * 1000 + patch
7979

8080
override def toString: String = s"$minorVersion.$patch${if isSnapshot then "-SNAPSHOT" else ""}"
81+
def isHigherThan(config: VersionConfig): Boolean =
82+
versionAsInt > config.versionAsInt
8183
end VersionConfig
8284

8385
object VersionConfig:

‎03-worker/src/main/scala/camundala/worker/Handler.scala

Lines changed: 38 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,30 @@ import sttp.model.Uri.QuerySegment
99
import sttp.model.{Method, Uri}
1010
import scala.reflect.ClassTag
1111

12+
trait WorkerHandler:
13+
def worker: Worker[?, ?, ?]
14+
def topic: String
15+
16+
def applicationName: String
17+
def registerHandler(): Unit
18+
def registerHandler( register: => Unit): Unit =
19+
val appPackageName = applicationName.replace("-", ".")
20+
val testMode = sys.env.get("WORKER_TEST_MODE").contains("true") // did not work with lazy val
21+
if testMode || getClass.getName.startsWith(appPackageName)
22+
then
23+
register
24+
logger.info(s"Worker registered: $topic -> ${worker.getClass.getSimpleName}")
25+
logger.debug(prettyString(worker))
26+
else
27+
logger.info(
28+
s"Worker NOT registered: $topic -> ${worker.getClass.getSimpleName} (class starts not with $appPackageName)"
29+
)
30+
end if
31+
end registerHandler
32+
33+
protected lazy val logger: WorkerLogger
34+
end WorkerHandler
35+
1236
/** handler for Custom Validation (next to the automatic Validation of the In Object.
1337
*
1438
* For example if one of two optional variables must exist.
@@ -125,11 +149,11 @@ case class ServiceHandler[
125149
val rRequest = runnableRequest(inputObject)
126150
for
127151
optWithServiceMock <- withServiceMock(rRequest, inputObject)
128-
output <- handleMocking(optWithServiceMock, rRequest).getOrElse(
129-
summon[EngineRunContext]
130-
.sendRequest[ServiceIn, ServiceOut](rRequest)
131-
.flatMap(out => outputMapper(out, inputObject))
132-
)
152+
output <- handleMocking(optWithServiceMock, rRequest).getOrElse(
153+
summon[EngineRunContext]
154+
.sendRequest[ServiceIn, ServiceOut](rRequest)
155+
.flatMap(out => outputMapper(out, inputObject))
156+
)
133157
yield output
134158
end for
135159
end runWork
@@ -157,13 +181,17 @@ case class ServiceHandler[
157181
case (_, Some(json)) =>
158182
(for
159183
mockedResponse <- decodeMock[MockedServiceResponse[ServiceOut]](json)
160-
out <- handleServiceMock(mockedResponse, runnableRequest, in)
184+
out <- handleServiceMock(mockedResponse, runnableRequest, in)
161185
yield out)
162186
.map(Some.apply)
163-
case (true, _) =>
164-
handleServiceMock(dynamicServiceOutMock.map(_(in)).getOrElse(defaultServiceOutMock), runnableRequest, in)
187+
case (true, _) =>
188+
handleServiceMock(
189+
dynamicServiceOutMock.map(_(in)).getOrElse(defaultServiceOutMock),
190+
runnableRequest,
191+
in
192+
)
165193
.map(Some.apply)
166-
case _ =>
194+
case _ =>
167195
Right(None)
168196

169197
end withServiceMock
@@ -200,7 +228,7 @@ case class ServiceHandler[
200228
mockedResponse match
201229
case MockedServiceResponse(_, Right(body), headers) =>
202230
mapBodyOutput(body, headers, in)
203-
case MockedServiceResponse(status, Left(body), _) =>
231+
case MockedServiceResponse(status, Left(body), _) =>
204232
Left(
205233
ServiceRequestError(
206234
status,

‎03-worker/src/main/scala/camundala/worker/WorkerDsl.scala

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@ trait WorkerDsl[In <: Product: InOutCodec, Out <: Product: InOutCodec]
1515

1616
// needed that it can be called from CSubscriptionPostProcessor
1717
def worker: Worker[In, Out, ?]
18-
1918
def topic: String = worker.topic
2019

2120
def runWorkFromWorker(in: In)(using EngineRunContext): Option[Either[RunWorkError, Out]] =

‎03-worker/src/main/scala/camundala/worker/exports.scala

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ import camundala.domain.*
66
import camundala.worker.CamundalaWorkerError.*
77
import io.circe.*
88

9+
import java.util.Date
10+
911
export sttp.model.Uri.UriContext
1012
export sttp.model.Method
1113
export sttp.model.Uri
@@ -168,3 +170,11 @@ end CamundalaWorkerError
168170

169171
def niceClassName(clazz: Class[?]) =
170172
clazz.getName.split("""\$""").head
173+
174+
def printTimeOnConsole(start: Date) =
175+
val time = new Date().getTime - start.getTime
176+
val color = if time > 1000 then Console.YELLOW_B
177+
else if time > 250 then Console.MAGENTA
178+
else Console.BLACK
179+
s"($color$time ms${Console.RESET})"
180+
end printTimeOnConsole

‎04-worker-c7spring/src/main/scala/camundala/camunda7/worker/C7WorkerHandler.scala

Lines changed: 3 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ import scala.util.{Failure, Success}
1818
/** To avoid Annotations (Camunda Version specific), we extend ExternalTaskHandler for required
1919
* parameters.
2020
*/
21-
trait C7WorkerHandler extends camunda.ExternalTaskHandler:
21+
trait C7WorkerHandler extends camunda.ExternalTaskHandler, WorkerHandler:
2222

2323
@Value("${spring.application.name}")
2424
var applicationName: String = scala.compiletime.uninitialized
@@ -29,9 +29,6 @@ trait C7WorkerHandler extends camunda.ExternalTaskHandler:
2929
@Autowired()
3030
protected var externalTaskClient: ExternalTaskClient = scala.compiletime.uninitialized
3131

32-
def worker: Worker[?, ?, ?]
33-
def topic: String
34-
3532
override def execute(
3633
externalTask: camunda.ExternalTask,
3734
externalTaskService: camunda.ExternalTaskService
@@ -43,27 +40,17 @@ trait C7WorkerHandler extends camunda.ExternalTaskHandler:
4340
)
4441
executeWorker(externalTaskService)(using externalTask)
4542
logger.info(
46-
s"Worker: ${externalTask.getTopicName} (${externalTask.getId}) ended ${printTime(startDate)} > ${externalTask.getBusinessKey}"
43+
s"Worker: ${externalTask.getTopicName} (${externalTask.getId}) ended ${printTimeOnConsole(startDate)} > ${externalTask.getBusinessKey}"
4744
)
4845
end execute
4946

5047
@PostConstruct
5148
def registerHandler(): Unit =
52-
val appPackageName = applicationName.replace("-", ".")
53-
val testMode = sys.env.get("WORKER_TEST_MODE").contains("true") // did not work with lazy val
54-
if testMode || getClass.getName.startsWith(appPackageName)
55-
then
49+
registerHandler:
5650
externalTaskClient
5751
.subscribe(topic)
5852
.handler(this)
5953
.open()
60-
logger.info(s"Worker registered: $topic -> ${worker.getClass.getSimpleName}")
61-
logger.debug(prettyString(worker))
62-
else
63-
logger.info(
64-
s"Worker NOT registered: $topic -> ${worker.getClass.getSimpleName} (class starts not with $appPackageName)"
65-
)
66-
end if
6754
end registerHandler
6855

6956
private def executeWorker(
@@ -94,14 +81,6 @@ trait C7WorkerHandler extends camunda.ExternalTaskHandler:
9481
end try
9582
end executeWorker
9683

97-
private def printTime(start: Date) =
98-
val time = new Date().getTime - start.getTime
99-
val color = if time > 1000 then Console.YELLOW_B
100-
else if time > 250 then Console.MAGENTA
101-
else Console.BLACK
102-
s"($color$time ms${Console.RESET})"
103-
end printTime
104-
10584
extension (externalTaskService: camunda.ExternalTaskService)
10685

10786
private def handleSuccess(

‎04-worker-c7spring/src/main/scala/camundala/camunda7/worker/exports.scala

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
package camundala
22
package camunda7.worker
33

4-
import camundala.worker.*
54
import org.camunda.bpm.client.task.ExternalTask
65
import sttp.client3.{HttpClientSyncBackend, Identity, SttpBackend}
76

‎04-worker-c8zio/src/main/scala/camundala/worker/c8zio/C8Worker.scala

Lines changed: 101 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,108 @@
11
package camundala.worker.c8zio
22

3-
import camundala.worker.JobWorker
3+
import camundala.domain.*
4+
import camundala.worker.{CamundalaWorkerError, JobWorker, Worker, printTimeOnConsole}
45
import io.camunda.zeebe.client.api.response.ActivatedJob
56
import io.camunda.zeebe.client.api.worker.{JobClient, JobHandler}
7+
import zio.*
68

7-
trait C8Worker extends JobWorker, JobHandler:
8-
def handle(client: JobClient, job: ActivatedJob): Unit =
9-
println(s"Handling Job: ${job}")
10-
client.newCompleteCommand(job.getKey).send().join()
9+
import java.util.Date
10+
11+
trait C8Worker[In: InOutDecoder, Out: InOutEncoder] extends JobWorker, JobHandler:
12+
def topic: String
13+
14+
lazy val runtime = Runtime.default
15+
16+
import cats.data.ValidatedNel
17+
import cats.implicits.*
18+
def handle(client: JobClient, job: ActivatedJob): Unit =
19+
Unsafe.unsafe:
20+
implicit unsafe =>
21+
runtime.unsafe.run(
22+
(for
23+
startDate <- ZIO.succeed(new Date())
24+
json <- ZIO.fromEither(parser.parse(job.getVariables))
25+
in <- ZIO.fromEither(
26+
customDecodeAccumulating[In](json.hcursor)
27+
)
28+
businessKey <-
29+
ZIO.fromEither(json.as[BusinessKey].map(_.businessKey.getOrElse("no businessKey")))
30+
_ <- Console.printLine(
31+
s"Worker: ${job.getType} (${job.getWorker}) started > $businessKey"
32+
)
33+
_ <- Console.printLine(s"IN: $in")
34+
_ <- ZIO.attempt(client.newCompleteCommand(job.getKey).send().join())
35+
_ <-
36+
Console.printLine(
37+
s"Worker: ${job.getType} (${job.getWorker}) ended ${printTimeOnConsole(startDate)} > $businessKey"
38+
)
39+
yield ())
40+
).getOrThrow()
41+
42+
def handle2(client: JobClient, job: ActivatedJob): Unit =
43+
Unsafe.unsafe { implicit unsafe =>
44+
runtime.unsafe.run(
45+
for
46+
businessKey <-
47+
ZIO.succeed(Option(job.getVariable("businessKey")).getOrElse("no businessKey"))
48+
_ <- Console.printLine(
49+
s"Worker: ${job.getType} (${job.getWorker}) started > $businessKey"
50+
)
51+
json <- ZIO.fromEither(parser.parse(job.getVariables))
52+
53+
_ <- Console.printLine(
54+
s"Worker: ${job.getType} (${job.getWorker}) started > $businessKey"
55+
)
56+
startDate <- ZIO.succeed(new Date())
57+
json <- ZIO.fromEither(parser.parse(job.getVariables))
58+
_ <- Console.printLine(
59+
s"Worker: ${job.getType} (${job.getWorker}) started > $businessKey"
60+
)
61+
_ <- handleJob(client, job)
62+
_ <-
63+
Console.printLine(
64+
s"Worker: ${job.getType} (${job.getWorker}) ended ${printTimeOnConsole(startDate)} > $businessKey"
65+
)
66+
_ <- ZIO.attempt(client.newCompleteCommand(job.getKey).send().join())
67+
yield ()
68+
).getOrThrowFiberFailure()
69+
}
70+
71+
private def handleJob(
72+
client: JobClient,
73+
job: ActivatedJob
74+
): ZIO[Any, Any, Any] = Console.printLine("handleJob")
75+
/* val variablesExtractor = ProcessVariablesExtractor(job)
76+
for
77+
tryProcessVariables <- variablesExtractor.extract(worker.variableNames ++ worker.inConfigVariableNames)
78+
tryGeneralVariables <- processExtractor.extractGeneral())
79+
filteredOut <- worker.executor(using EngineRunContext(engineContext, tryGeneralVariables)).execute(tryProcessVariables)
80+
val tryProcessVariables =
81+
ProcessVariablesExtractor(job).extract(worker.variableNames ++ worker.inConfigVariableNames)
82+
val tryGeneralVariables = ProcessVariablesExtractor.extractGeneral()
83+
try
84+
(for
85+
generalVariables <- tryGeneralVariables
86+
context = EngineRunContext(engineContext, generalVariables)
87+
filteredOut <-
88+
worker.executor(using context).execute(tryProcessVariables)
89+
yield externalTaskService.handleSuccess(filteredOut, generalVariables.manualOutMapping) //
90+
).left.map { ex =>
91+
externalTaskService.handleError(ex, tryGeneralVariables)
92+
}
93+
catch // safety net
94+
case ex: Throwable =>
95+
ex.printStackTrace()
96+
externalTaskService.handleError(
97+
UnexpectedError(errorMsg =
98+
s"We caught an UnhandledException: ${ex.getMessage}\n - check the Workers Log."
99+
),
100+
tryGeneralVariables
101+
)
102+
end try
103+
end handleJob*/
104+
case class BusinessKey(businessKey: Option[String])
105+
object BusinessKey:
106+
given InOutCodec[BusinessKey] = deriveInOutCodec
11107

12108
end C8Worker

‎04-worker-c8zio/src/main/scala/camundala/worker/c8zio/C8WorkerClient.scala

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,9 @@ import zio.{Console, ZIO}
77

88
import java.net.URI
99

10-
object C8WorkerClient extends WorkerClient[C8Worker]:
10+
object C8WorkerClient extends WorkerClient[C8Worker[?,?]]:
1111

12-
def runWorkers(workers: Set[C8Worker]): ZIO[Any, Any, Any] =
12+
def runWorkers(workers: Set[C8Worker[?,?]]): ZIO[Any, Any, Any] =
1313
Console.printLine(s"Starting Zeebe Worker Client: ${workers}") *>
1414
ZIO.acquireReleaseWith(zeebeClient)(_.closeClient()): client =>
1515
for
@@ -23,7 +23,7 @@ object C8WorkerClient extends WorkerClient[C8Worker]:
2323
_ <- server.join
2424
yield ()
2525

26-
private def registerWorker(worker: C8Worker, client: ZeebeClient) =
26+
private def registerWorker(worker: C8Worker[?,?], client: ZeebeClient) =
2727
Console.printLine("Registering Worker: " + worker.topic) *>
2828
ZIO.attempt(client
2929
.newWorker()

‎04-worker-c8zio/src/main/scala/camundala/worker/c8zio/C8WorkerHandler.scala

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,10 @@
1-
package camundala.worker.c8zio
1+
package camundala.worker
2+
package c8zio
23

34
/** To avoid Annotations (Camunda Version specific), we extend ExternalTaskHandler for required
45
* parameters.
56
*/
6-
trait C8WorkerHandler:
7+
trait C8WorkerHandler extends WorkerHandler:
78

89

910
end C8WorkerHandler

‎05-examples/demos/03-worker/src/main/scala/camundala/examples/demos/newWorker/CompanyWorkerHandler.scala

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -7,28 +7,31 @@ import camundala.worker.c8zio.C8Worker
77

88
import scala.reflect.ClassTag
99

10-
trait CompanyWorkerHandler extends C7WorkerHandler, C8Worker
10+
trait CompanyWorkerHandler[
11+
In <: Product: InOutCodec,
12+
Out <: Product: InOutCodec
13+
] extends C7WorkerHandler, C8Worker[In, Out]
1114

1215
trait CompanyValidationWorkerDsl[
1316
In <: Product: InOutCodec
14-
] extends CompanyWorkerHandler, ValidationWorkerDsl[In]
17+
] extends CompanyWorkerHandler[In, NoOutput], ValidationWorkerDsl[In]
1518

1619
trait CompanyInitWorkerDsl[
1720
In <: Product: InOutCodec,
1821
Out <: Product: InOutCodec,
1922
InitIn <: Product: InOutCodec,
2023
InConfig <: Product: InOutCodec
21-
] extends CompanyWorkerHandler, InitWorkerDsl[In, Out, InitIn, InConfig]
24+
] extends CompanyWorkerHandler[In, Out], InitWorkerDsl[In, Out, InitIn, InConfig]
2225

2326
trait CompanyCustomWorkerDsl[
2427
In <: Product: InOutCodec,
2528
Out <: Product: InOutCodec
26-
] extends CompanyWorkerHandler, CustomWorkerDsl[In, Out]
29+
] extends CompanyWorkerHandler[In, Out], CustomWorkerDsl[In, Out]
2730

2831

2932
trait CompanyServiceWorkerDsl[
3033
In <: Product: InOutCodec,
3134
Out <: Product: InOutCodec,
3235
ServiceIn: InOutEncoder,
3336
ServiceOut: InOutDecoder: ClassTag
34-
] extends CompanyWorkerHandler, ServiceWorkerDsl[In, Out, ServiceIn, ServiceOut]
37+
] extends CompanyWorkerHandler[In, Out], ServiceWorkerDsl[In, Out, ServiceIn, ServiceOut]

0 commit comments

Comments
 (0)