@@ -60,7 +60,8 @@ void chat_bot() {
60
60
// .chatMemoryProvider(memoryId -> MessageWindowChatMemory.withMaxMessages(10))
61
61
.outputName ("conversation" )
62
62
.build ());
63
- BlockingQueue <CloudEvent > publishedEvents = new LinkedBlockingQueue <>();
63
+ BlockingQueue <CloudEvent > replyEvents = new LinkedBlockingQueue <>();
64
+ BlockingQueue <CloudEvent > finishedEvents = new LinkedBlockingQueue <>();
64
65
65
66
// 1. listen to an event containing `message` key in the body
66
67
// 2. if contains, call the agent, if not end the workflow
@@ -71,7 +72,13 @@ void chat_bot() {
71
72
t ->
72
73
t .listen (
73
74
l ->
74
- l .until (message -> "" .equals (message .get ("userInput" )), Map .class )
75
+ l .until (
76
+ message ->
77
+ !message
78
+ .getOrDefault ("userInput" , "" )
79
+ .toString ()
80
+ .isEmpty (),
81
+ Map .class )
75
82
.any (
76
83
c ->
77
84
c .with (event -> event .type ("org.acme.chatbot.request" )))
@@ -87,7 +94,16 @@ void chat_bot() {
87
94
e ->
88
95
e .type (
89
96
"org.acme.chatbot.reply" ))))))
90
- .emit (emit -> emit .event (e -> e .type ("org.acme.chatbot.finished" ))))
97
+ .emit (
98
+ emit ->
99
+ emit .when (
100
+ message ->
101
+ message
102
+ .getOrDefault ("userInput" , "" )
103
+ .toString ()
104
+ .isEmpty (),
105
+ Map .class )
106
+ .event (e -> e .type ("org.acme.chatbot.finished" ))))
91
107
.build ();
92
108
93
109
try (WorkflowApplication app = WorkflowApplication .builder ().build ()) {
@@ -98,7 +114,7 @@ void chat_bot() {
98
114
new EventFilter ()
99
115
.withWith (new EventProperties ().withType ("org.acme.chatbot.reply" )),
100
116
app ),
101
- ce -> publishedEvents .add ((CloudEvent ) ce ));
117
+ ce -> replyEvents .add ((CloudEvent ) ce ));
102
118
103
119
app .eventConsumer ()
104
120
.register (
@@ -107,7 +123,7 @@ void chat_bot() {
107
123
new EventFilter ()
108
124
.withWith (new EventProperties ().withType ("org.acme.chatbot.finished" )),
109
125
app ),
110
- ce -> publishedEvents .add ((CloudEvent ) ce ));
126
+ ce -> finishedEvents .add ((CloudEvent ) ce ));
111
127
112
128
final WorkflowInstance waitingInstance =
113
129
app .workflowDefinition (listenWorkflow ).instance (Map .of ());
@@ -118,12 +134,12 @@ void chat_bot() {
118
134
119
135
// Publish the event
120
136
app .eventPublisher ().publish (newMessageEvent ("Hello World!" ));
121
- CloudEvent reply = publishedEvents .poll (60 , TimeUnit .SECONDS );
137
+ CloudEvent reply = replyEvents .poll (60 , TimeUnit .SECONDS );
122
138
assertNotNull (reply );
123
139
124
140
// Empty message completes the workflow
125
141
app .eventPublisher ().publish (newMessageEvent ("" ));
126
- CloudEvent finished = publishedEvents .poll (60 , TimeUnit .SECONDS );
142
+ CloudEvent finished = finishedEvents .poll (60 , TimeUnit .SECONDS );
127
143
assertNotNull (finished );
128
144
129
145
assertThat (runningModel ).isCompleted ();
@@ -158,7 +174,7 @@ void mixed_workflow() {
158
174
AgenticServices .agentBuilder (Agents .ChatBot .class )
159
175
.chatModel (Models .BASE_MODEL )
160
176
.chatMemoryProvider (memoryId -> MessageWindowChatMemory .withMaxMessages (10 ))
161
- .outputName ("message " )
177
+ .outputName ("userInput " )
162
178
.build ());
163
179
164
180
final Workflow mixedWorkflow =
@@ -170,7 +186,7 @@ void mixed_workflow() {
170
186
callJ .function (
171
187
input -> {
172
188
System .out .println (input );
173
- return Map .of ("message " , input );
189
+ return Map .of ("userInput " , input );
174
190
},
175
191
String .class ))
176
192
.agent (chatBot )
0 commit comments