<commit_before>import threado
import disco
import uuid
from xmlcore import Element, Query
from core import STANZA_NS, XMPPError
from jid import JID

MUC_NS = "http://jabber.org/protocol/muc"
USER_NS = MUC_NS + "#user"
OWNER_NS = MUC_NS + "#owner"
ROOMS_NODE = "http://jabber.org/protocol/muc#rooms"

class MUCError(XMPPError):
    pass

def gen_random():
    return uuid.uuid4().hex[:8]

def parse_presence(elements, own_jid):
    presences = elements.named("presence", STANZA_NS)

    for presence in presences.with_attrs("from"):
        other_jid = JID(presence.get_attr("from"))
        if other_jid.bare() != own_jid.bare():
            continue

        type = presence.get_attr("type", None)
        if other_jid == own_jid and type == "error":
            raise MUCError("could not join room", presence.children("error"))

        x = presence.children("x", USER_NS)
        codes = set()
        for status in x.children("status").with_attrs("code"):
            codes.add(status.get_attr("code"))

        for item in x.children("item").with_attrs("affiliation", "role"):
            jid = item.get_attr("jid", None)
            if jid is not None:
                jid = JID(jid)

            affiliation = item.get_attr("affiliation")
            role = item.get_attr("role")
            participant = MUCParticipant(other_jid, affiliation, role,
                                         presence.children(), jid)
            return participant, codes
    return None

class MUCRoom(threado.GeneratorStream):
    def __init__(self, muc, xmpp, stream, jid):
        threado.GeneratorStream.__init__(self)

        self.room_jid = JID(jid).bare()
        self.nick_jid = JID(jid)
        self.exit_done = threado.Channel()
        self.participants = list()

        self.muc = muc
        self.xmpp = xmpp
        self.stream = stream

    @threado.stream_fast
    def _join(inner, self, password=None, history=False):
        x = Element("x", xmlns=MUC_NS)
        if password is not None:
            password_element = Element("password")
            password_element.text = password
            x.add(password_element)
        if not history:
            history_element = Element("history", maxstanzas="0")
            x.add(history_element)
        self.xmpp.core.presence(x, to=JID(self.nick_jid))

        while True:
            yield inner, self.stream
            list(inner)

            for element in self.stream:
                parsed = parse_presence(element, self.nick_jid)
                if parsed is None:
                    continue

                participant, codes = parsed
                self.participants.append(participant)

                if "201" in codes:
                    submit = Element("x", xmlns="jabber:x:data", type="submit")
                    query = Element("query", xmlns=OWNER_NS)
                    query.add(submit)
                    self.xmpp.core.iq_set(query, to=self.room_jid)

                if participant.name == self.nick_jid or "110" in codes:
                    nick_jid = participant.name
                    self.start()
                    return

    @threado.stream
    def exit(inner, self, reason=None):
        if reason is not None:
            status = Element("status")
            status.text = reason
            self.xmpp.core.presence(status, to=self.nick_jid, type="unavailable")
        else:
            self.xmpp.core.presence(to=self.nick_jid, type="unavailable")

        yield self.exit_done

    @threado.stream_fast
    def _run(inner, self):
        exit_sent = False

        try:
            while True:
                yield inner, self.stream

                try:
                    for elements in inner:
                        if exit_sent:
                            continue
                        if elements is None:
                            elements = list()
                        attrs = dict(type="groupchat")
                        self.xmpp.core.message(self.room_jid, *elements, **attrs)
                        continue
                except threado.Finished:
                    if not exit_sent:
                        self.exit()
                        exit_send = True
                except:
                    if not exit_sent:
                        self.exit()
                        exit_send = True
                    raise

                for elements in self.stream:
                    for element in elements.with_attrs("from"):
                        from_jid = JID(element.get_attr("from"))
                        if from_jid.bare() != self.room_jid:
                            continue

                        inner.send(element)

                        for presence in element.named("presence"):
                            if presence.get_attr("type", None) != "unavailable":
                                continue
                            if from_jid == self.nick_jid:
                                return
                            for x in presence.children("x", USER_NS):
                                if x.children("status").with_attrs("code", "110"):
                                    return
        finally:
            self.muc._exit_room(self)
            self.exit_done.finish()

    def run(self):
        yield self.inner.sub(self._run())

class MUCParticipant(object):
    def __init__(self, name, affiliation, role, payload, jid=None):
        self.name = name
        self.affiliation = affiliation
        self.role = role
        self.jid = jid
        self.payload = payload

class MUC(object):
    def __init__(self, xmpp):
        self.xmpp = xmpp
        self.xmpp.disco.add_feature(MUC_NS)
        self.xmpp.disco.add_node(ROOMS_NODE, self._node_handler)
        self.rooms = dict()
        self.xmpp.add_listener(self._handler)

    def _handler(self, success, value):
        if success:
            bare = JID(value.get_attr("from")).bare()
            for room in self.rooms.get(bare, set()):
                room.stream.send(value)
        else:
            for bare, rooms in self.rooms.iteritems():
                for room in rooms:
                    room.stream.throw(*value)
            self.rooms.clear()<commit_msg>idiokit/muc.py: MUC implementation doesn't stream_fast anymore.<commit_after>import threado
import disco
import uuid
from xmlcore import Element, Query
from core import STANZA_NS, XMPPError
from jid import JID

MUC_NS = "http://jabber.org/protocol/muc"
USER_NS = MUC_NS + "#user"
OWNER_NS = MUC_NS + "#owner"
ROOMS_NODE = "http://jabber.org/protocol/muc#rooms"

class MUCError(XMPPError):
    pass

def gen_random():
    return uuid.uuid4().hex[:8]

def parse_presence(elements, own_jid):
    presences = elements.named("presence", STANZA_NS)

    for presence in presences.with_attrs("from"):
        other_jid = JID(presence.get_attr("from"))
        if other_jid.bare() != own_jid.bare():
            continue

        type = presence.get_attr("type", None)
        if other_jid == own_jid and type == "error":
            raise MUCError("could not join room", presence.children("error"))

        x = presence.children("x", USER_NS)
        codes = set()
        for status in x.children("status").with_attrs("code"):
            codes.add(status.get_attr("code"))

        for item in x.children("item").with_attrs("affiliation", "role"):
            jid = item.get_attr("jid", None)
            if jid is not None:
                jid = JID(jid)

            affiliation = item.get_attr("affiliation")
            role = item.get_attr("role")
            participant = MUCParticipant(other_jid, affiliation, role,
                                         presence.children(), jid)
            return participant, codes
    return None

class MUCRoom(threado.GeneratorStream):
    def __init__(self, muc, xmpp, stream, jid):
        threado.GeneratorStream.__init__(self)

        self.room_jid = JID(jid).bare()
        self.nick_jid = JID(jid)
        self.exit_done = threado.Channel()
        self.participants = list()

        self.muc = muc
        self.xmpp = xmpp
        self.stream = stream

    @threado.stream
    def _join(inner, self, password=<PASSWORD>, history=False):
        x = Element("x", xmlns=MUC_NS)
        if password is not None:
            password_element = Element("password")
            password_element.text = password
            x.add(password_element)
        if not history:
            history_element = Element("history", maxstanzas="0")
            x.add(history_element)
        self.xmpp.core.presence(x, to=JID(self.nick_jid))

        while True:
            element = yield inner, self.stream
            if inner.was_source:
                continue

            parsed = parse_presence(element, self.nick_jid)
            if parsed is None:
                continue

            participant, codes = parsed
            self.participants.append(participant)

            if "201" in codes:
                submit = Element("x", xmlns="jabber:x:data", type="submit")
                query = Element("query", xmlns=OWNER_NS)
                query.add(submit)
                self.xmpp.core.iq_set(query, to=self.room_jid)

            if participant.name == self.nick_jid or "110" in codes:
                nick_jid = participant.name
                self.start()
                return

    @threado.stream
    def exit(inner, self, reason=None):
        if reason is not None:
            status = Element("status")
            status.text = reason
            self.xmpp.core.presence(status, to=self.nick_jid, type="unavailable")
        else:
            self.xmpp.core.presence(to=self.nick_jid, type="unavailable")

        yield self.exit_done

    @threado.stream
    def _send(inner, self):
        while True:
            try:
                elements = yield inner
            except threado.Finished:
                self.exit()
                break
            except:
                self.exit()
                raise

            if elements is None:
                elements = list()

            attrs = dict(type="groupchat")
            self.xmpp.core.message(self.room_jid, *elements, **attrs)

        while True:
            try:
                elements = yield inner
            except threado.Finished:
                pass

    @threado.stream
    def _receive(inner, self):
        while True:
            elements = yield inner, self.stream
            if inner.was_source:
                continue

            for element in elements.with_attrs("from"):
                from_jid = JID(element.get_attr("from"))
                if from_jid.bare() != self.room_jid:
                    continue

                inner.send(element)

                for presence in element.named("presence"):
                    if presence.get_attr("type", None) != "unavailable":
                        continue
                    if from_jid == self.nick_jid:
                        return
                    for x in presence.children("x", USER_NS):
                        if x.children("status").with_attrs("code", "110"):
                            return

    def run(self):
        try:
            yield self.inner.sub(self._send() | self._receive())
        finally:
            self.muc._exit_room(self)
            self.exit_done.finish()

class MUCParticipant(object):
    def __init__(self, name, affiliation, role, payload, jid=None):
        self.name = name
        self.affiliation = affiliation
        self.role = role
        self.jid = jid
        self.payload = payload

class MUC(object):
    def __init__(self, xmpp):
        self.xmpp = xmpp
        self.xmpp.disco.add_feature(MUC_NS)
        self.xmpp.disco.add_node(ROOMS_NODE, self._node_handler)
        self.rooms = dict()
        self.xmpp.add_listener(self._handler)

    def _handler(self, success, value):
        if success:
            bare = JID(value.get_attr("from")).bare()
            for room in self.rooms.get(bare, set()):
                room.stream.send(value)
        else:
            for bare, rooms in self.rooms.iteritems():
                for room in rooms:
                    room.stream.throw(*value)
            self.rooms.clear()
{'fraction_non_alphanumeric': 0.10192071924805884, 'fraction_numerical': 0.001961585615038823, 'mean_word_length': 1.6319638631963864, 'pattern_counts': {'":': 4, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 8}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3984930', 'n_tokens_mistral': 3588, 'n_tokens_neox': 3409, 'n_words': 759}