@@ -157,12 +157,31 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
157157 def ' should not call onComplete after unsubscribe is called' () {
158158 given :
159159 def block = getBlock()
160- def observer = new TestObserver ()
160+ def observer = new TestObserver (new Observer () {
161+ private Subscription subscription
162+
163+ @Override
164+ void onSubscribe (final Subscription subscription ) {
165+ this . subscription = subscription
166+ }
167+
168+ @Override
169+ void onNext (final Object result ) {
170+ subscription. unsubscribe()
171+ }
172+
173+ @Override
174+ void onError (final Throwable e ) {
175+ }
176+
177+ @Override
178+ void onComplete () {
179+ }
180+ })
161181 observe(block). subscribe(observer)
162182
163183 when :
164184 observer. requestMore(1 )
165- observer. getSubscription(). unsubscribe()
166185
167186 then :
168187 observer. assertUnsubscribed()
@@ -174,12 +193,16 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
174193 given :
175194 def block = getBlock()
176195 def observer = new TestObserver (new Observer () {
196+ private Subscription subscription
197+
177198 @Override
178199 void onSubscribe (final Subscription subscription ) {
200+ this . subscription = subscription
179201 }
180202
181203 @Override
182204 void onNext (final Object result ) {
205+ subscription. unsubscribe()
183206 throw new MongoException (' Failure' )
184207 }
185208
@@ -195,13 +218,11 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
195218
196219 when :
197220 observer. requestMore(1 )
198- observer. getSubscription(). unsubscribe()
199221
200222 then :
201- observer. assertUnsubscribed()
202223 observer. assertReceivedOnNext([1 ])
203- observer. assertErrored ()
204- observer. assertTerminalEvent ()
224+ observer. assertUnsubscribed ()
225+ observer. assertNoTerminalEvent ()
205226 }
206227
207228
0 commit comments