You need to perform the subscription on a scheduler such as currentThread to avoid stack overflow. 

In addition, 

- it  is not advised to call subscribe from a constructor,
- the call from the constructor is not protected by a lock and there is a race condition between ```subscribed.set``` and unsubsctibeFromSource,
- even if using lock, subscribeToSource may be reentered before the subscribe returns and ```subscribed``` might not be set to true at all,
- you'd need to use MultipleAssignmentSubscription which does the correct lifecycle management for this operator (the original bug was a per-source shared composite instead of a per-observer composite plus the retry counter).