@transaction.atomic
    def do_sub(self, id_, name, silent, *params):
        """Subscribe the current thread to the specified publication."""
        try:
            pub = self.get_pub_by_name(name)
        except __HOLE__:
            if not silent:
                raise MeteorError(404, 'Subscription not found')
            return
        sub, created = Subscription.objects.get_or_create(
            connection_id=this.ws.connection.pk,
            sub_id=id_,
            user_id=getattr(this, 'user_id', None),
            defaults={
                'publication': pub.name,
                'params_ejson': ejson.dumps(params),
            },
        )
        this.subs.setdefault(sub.publication, set()).add(sub.pk)
        if not created:
            if not silent:
                this.send({'msg': 'ready', 'subs': [id_]})
            return
        # re-read from DB so we can get transaction ID (xmin)
        sub = Subscription.objects.extra(**XMIN).get(pk=sub.pk)
        for col, qs in self.sub_unique_objects(
                sub, params, pub, xmin__lte=sub.xmin,
        ):
            sub.collections.create(
                model_name=model_name(qs.model),
                collection_name=col.name,
            )
            if isinstance(col.model._meta.pk, AleaIdField):
                meteor_ids = None
            elif len([
                field
                for field
                in col.model._meta.local_fields
                if (
                    isinstance(field, AleaIdField)
                ) and (
                    field.unique
                ) and (
                    not field.null
                )
            ]) == 1:
                meteor_ids = None
            else:
                meteor_ids = get_meteor_ids(
                    qs.model, qs.values_list('pk', flat=True),
                )
            for obj in qs.select_related():
                payload = col.obj_change_as_msg(obj, ADDED, meteor_ids)
                this.send(payload)
        if not silent:
            this.send({'msg': 'ready', 'subs': [id_]})
dataset/ETHPy150Open django-ddp/django-ddp/dddp/api.py/DDP.do_sub