human
... via multicast
RxPY also has a multicast operator which operates on an ordinary Observable, multicasts that Observable by means of a particular Subject that you specify, applies a transformative function to each emission, and then emits those transformed values as its own ordinary Observable sequence.
Each subscription to this new Observable will trigger a new subscription to the underlying multicast Observable.
Following the RXJS example at reactive.io docu:
assistant
#Here is the Python program you requested.
rst(O.multicast)
# show actions on intermediate subject:
show = False

def emit(obs):
    'instead of range we allow some logging:'
    for i in (1, 2):
        v = rand()
        log('emitting', v)
        obs.on_next(v)
    log('complete')
    obs.on_completed()
    

class MySubject:
    def __init__(self):
        self.rx_subj = Subject()
        if show:
            log('New Subject %s created' % self)

        
    def __str__(self):
        return str(hash(self))[-4:]
    
    def __getattr__(self, a):
        'called at any attr. access, logging it'
        if not a.startswith('__') and show:
            log('RX called', a, 'on MySub\n')
        return getattr(self.rx_subj, a)
        
        
subject1 = MySubject()
subject2 = MySubject()

source = O.create(emit).multicast(subject2)

# a "subscription" *is* a disposable
# (the normal d we return all the time):
d, observer  = subs(source, return_subscriber=True)
ds1 = subject1.subscribe(observer)
ds2 = subject2.subscribe(observer)
print ('we have now 3 subscriptions, only two will see values.')
print('start multicast stream (calling connect):')
connected = source.connect()
d.dispose()

rst(O.let)
# show actions on intermediate subject:
show = True

def emit(obs):
    'instead of range we allow some logging:'
    v = rand()
    log('emitting', v)
    obs.on_next(v)
    log('complete')
    obs.on_completed()
    
source = O.create(emit)

# following the RXJS example:
header("without let")
d = subs(source.concat(source))
d = subs(source.concat(source))

header("now with let")
d = subs(source.let(lambda o: o.concat(o)))
d = subs(source.let(lambda o: o.concat(o)))
# TODO: Not understood:
# "This operator allows for a fluent style of writing queries that use the same sequence multiple times."
# ... I can't verify this, the source sequence is not duplicated but called every time like a cold obs.