1 /*
2 * *************************************************************************************************************************************************************
3 *
4 * TheseFoolishThings: Miscellaneous utilities
5 * http://tidalwave.it/projects/thesefoolishthings
6 *
7 * Copyright (C) 2009 - 2025 by Tidalwave s.a.s. (http://tidalwave.it)
8 *
9 * *************************************************************************************************************************************************************
10 *
11 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
12 * You may obtain a copy of the License at
13 *
14 * http://www.apache.org/licenses/LICENSE-2.0
15 *
16 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
17 * CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
18 *
19 * *************************************************************************************************************************************************************
20 *
21 * git clone https://bitbucket.org/tidalwave/thesefoolishthings-src
22 * git clone https://github.com/tidalwave-it/thesefoolishthings-src
23 *
24 * *************************************************************************************************************************************************************
25 */
26 package it.tidalwave.actor.impl;
27
28 import java.lang.reflect.Method;
29 import javax.annotation.Nonnull;
30 import it.tidalwave.actor.annotation.Message;
31 import it.tidalwave.actor.spi.ActorActivatorStats;
32 import it.tidalwave.messagebus.MessageBus;
33 import lombok.RequiredArgsConstructor;
34 import lombok.extern.slf4j.Slf4j;
35
36 /***************************************************************************************************************************************************************
37 *
38 * @author Fabrizio Giudici
39 *
40 **************************************************************************************************************************************************************/
41 @RequiredArgsConstructor @Slf4j
42 class MessageListenerAdapter<T> implements MessageBus.Listener<T>
43 {
44 @Nonnull
45 private final Object owner;
46
47 @Nonnull
48 private final Method method;
49
50 @Nonnull
51 private final ExecutorWithPriority executor;
52
53 @Nonnull
54 private final ActorActivatorStats stats;
55
56 @Override
57 public void notify (@Nonnull final T message)
58 {
59 log.trace("notify({})", message);
60 final var collaboration = DefaultCollaboration.getCollaboration(message);
61 collaboration.registerPendingMessage(message);
62 stats.changePendingMessageCount(+1);
63
64 final Runnable messageWorker = () ->
65 {
66 collaboration.bindToCurrentThread();
67 collaboration.unregisterPendingMessage(message);
68 stats.changePendingMessageCount(-1);
69
70 try
71 {
72 stats.incrementInvocationCount();
73 method.invoke(owner, message);
74 }
75 catch (Throwable t)
76 {
77 stats.incrementInvocationErrorCount();
78 log.error("Error calling {} with {}", method, message.getClass());
79 log.error("", t);
80 }
81 finally
82 {
83 stats.incrementSuccessfulInvocationCount();
84 collaboration.unbindFromCurrentThread();
85 }
86 };
87
88 if (message.getClass().getAnnotation(Message.class).outOfBand())
89 {
90 executor.executeWithPriority(messageWorker);
91 }
92 else
93 {
94 executor.execute(messageWorker);
95 }
96 }
97 }
98
99