'Invalidates the cache and adds it to the cache stream so slaves
will know to invalidate their caches.
This should only be used to invalidate caches where slaves won\'t
otherwise know from other replication streams that the cache should
be invalidated.'

def _invalidate_cache_and_stream(self, txn, cache_func, keys):

    txn.call_after(cache_func.invalidate, keys)
    if isinstance(self.database_engine, PostgresEngine):
        ctx = self._cache_id_gen.get_next()
        stream_id = ctx.__enter__()
        txn.call_finally(ctx.__exit__, None, None, None)
        txn.call_after(self.hs.get_notifier().on_new_replication_data)
        self._simple_insert_txn(txn, table='cache_invalidation_stream', values={'stream_id': stream_id, 'cache_func': cache_func.__name__, 'keys': list(keys), 'invalidation_ts': self.clock.time_msec()})
