17
17
18
18
import static org .assertj .core .api .Assertions .assertThat ;
19
19
import static org .mockito .Mockito .inOrder ;
20
+ import static org .mockito .Mockito .timeout ;
20
21
import static org .mockito .Mockito .verify ;
21
22
22
23
import com .datastax .oss .driver .api .core .CqlSession ;
33
34
import com .datastax .oss .driver .api .testinfra .session .SessionUtils ;
34
35
import com .datastax .oss .driver .api .testinfra .simulacron .SimulacronRule ;
35
36
import com .datastax .oss .driver .categories .ParallelizableTests ;
37
+ import com .datastax .oss .driver .shaded .guava .common .util .concurrent .Uninterruptibles ;
36
38
import com .datastax .oss .simulacron .common .cluster .ClusterSpec ;
37
39
import edu .umd .cs .findbugs .annotations .NonNull ;
38
40
import java .util .Collections ;
41
+ import java .util .concurrent .CountDownLatch ;
42
+ import java .util .concurrent .TimeUnit ;
39
43
import org .junit .ClassRule ;
40
44
import org .junit .Test ;
41
45
import org .junit .experimental .categories .Category ;
@@ -90,6 +94,9 @@ public void should_inject_session_in_listeners() throws Exception {
90
94
.build ())
91
95
.build ()) {
92
96
97
+ // These NodeStateListeners are wrapped with SafeInitNodeStateListener which delays #onUp
98
+ // callbacks until #onSessionReady is called, these will all happen during session
99
+ // initialization
93
100
InOrder inOrder1 = inOrder (nodeListener1 );
94
101
inOrder1 .verify (nodeListener1 ).onSessionReady (session );
95
102
inOrder1 .verify (nodeListener1 ).onUp (nodeCaptor1 .capture ());
@@ -104,20 +111,29 @@ public void should_inject_session_in_listeners() throws Exception {
104
111
assertThat (nodeCaptor2 .getValue ().getEndPoint ())
105
112
.isEqualTo (SIMULACRON_RULE .getContactPoints ().iterator ().next ());
106
113
107
- verify (schemaListener1 ).onSessionReady (session );
108
- verify (schemaListener2 ).onSessionReady (session );
114
+ // SchemaChangeListener#onSessionReady is called asynchronously from AdminExecutor so we may
115
+ // have to wait a little
116
+ verify (schemaListener1 , timeout (500 ).times (1 )).onSessionReady (session );
117
+ verify (schemaListener2 , timeout (500 ).times (1 )).onSessionReady (session );
109
118
119
+ // Request tracker #onSessionReady is called synchronously during session initialization
110
120
verify (requestTracker1 ).onSessionReady (session );
111
121
verify (requestTracker2 ).onSessionReady (session );
112
122
113
123
assertThat (MyNodeStateListener .onSessionReadyCalled ).isTrue ();
114
124
assertThat (MyNodeStateListener .onUpCalled ).isTrue ();
115
125
116
- assertThat (MySchemaChangeListener .onSessionReadyCalled ).isTrue ();
126
+ // SchemaChangeListener#onSessionReady is called asynchronously from AdminExecutor so we may
127
+ // have to wait a little
128
+ assertThat (
129
+ Uninterruptibles .awaitUninterruptibly (
130
+ MySchemaChangeListener .onSessionReadyLatch , 500 , TimeUnit .MILLISECONDS ))
131
+ .isTrue ();
117
132
118
133
assertThat (MyRequestTracker .onSessionReadyCalled ).isTrue ();
119
134
}
120
135
136
+ // CqlSession#close waits for all listener close methods to be called
121
137
verify (nodeListener1 ).close ();
122
138
verify (nodeListener2 ).close ();
123
139
@@ -163,14 +179,14 @@ public void close() {
163
179
164
180
public static class MySchemaChangeListener extends SchemaChangeListenerBase {
165
181
166
- private static volatile boolean onSessionReadyCalled = false ;
182
+ private static CountDownLatch onSessionReadyLatch = new CountDownLatch ( 1 ) ;
167
183
private static volatile boolean closeCalled = false ;
168
184
169
185
public MySchemaChangeListener (@ SuppressWarnings ("unused" ) DriverContext ignored ) {}
170
186
171
187
@ Override
172
188
public void onSessionReady (@ NonNull Session session ) {
173
- onSessionReadyCalled = true ;
189
+ onSessionReadyLatch . countDown () ;
174
190
}
175
191
176
192
@ Override
0 commit comments