<commit_before># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import logging
import sched
import socket
import ssl
import uuid
try:
    from base64 import encodestring as encodebytes
except ImportError:
    # who the hell OKed this backwards incompatible change??
    from base64 import encodebytes
from socket import error as socket_error
from time import time

from wampy.backends import async_adapter
from wampy.backends.errors import WampyTimeOut
from wampy.config.defaults import heartbeat, heartbeat_timeout
from wampy.constants import (
    WEBSOCKET_SUBPROTOCOLS, WEBSOCKET_VERSION,
)
from wampy.errors import (
    IncompleteFrameError, NoFrameReturnedError, WampProtocolError, WampyError,
)
from wampy.interfaces import Transport
from wampy.mixins import ParseUrlMixin
from wampy.serializers import json_serialize

from . frames import FrameFactory, Ping, Pong, Text

logger = logging.getLogger(__name__)


class WebSocket(Transport, ParseUrlMixin):

    def __init__(self, server_url, ipv=4):

        self.url = server_url
        self.ipv = ipv

        self.host = None
        self.port = None
        self.resource = None

        self.parse_url()
        self.websocket_location = self.resource
        self.key = encodebytes(uuid.uuid4().bytes).decode('utf-8').strip()
        self.socket = None
        self.connected = False

        self._first_pinged_at = None
        self._pinged_at = None
        self._pong_pointer = None

        self.pongs = async_adapter.queue()
        self.missed_pongs = 0
        self.is_pinging = False

    def connect(self, upgrade=True):
        # TCP connection
        self._connect()
        self._handshake(upgrade=upgrade)

        if heartbeat > 0:
            self.start_pinging()
        return self

    def disconnect(self):
        logger.warning("disconnecting from %s", self.url)
        if self.socket:
            try:
                self.socket.shutdown(socket.SHUT_RDWR)
            except socket.error:
                pass

            self.socket.close()

    def send(self, message):
        frame = Text(payload=json_serialize(message))
        websocket_message = frame.frame
        self._send_raw(websocket_message)

    def _send_raw(self, websocket_message):
        logger.debug('send raw: %s', websocket_message)
        self.socket.sendall(websocket_message)

    def receive(self, bufsize=1):
        frame = None
        received_bytes = bytearray()
        conn = self.socket

        while True:

            try:
                bytes_ = conn.recv(bufsize)
            except socket.timeout as e:
                message = str(e)
                raise ConnectionError('timeout: "{}"'.format(message))
            except OSError:
                logger.info("socket connection lost? Bad File Descriptor?")
                break
            except Exception as exc:
                raise ConnectionError('Connection lost: "{}"'.format(exc))
            if not bytes_:
                break

            received_bytes.extend(bytes_)

            try:
                frame = FrameFactory.from_bytes(received_bytes)
            except IncompleteFrameError as exc:
                bufsize = exc.required_bytes
            else:
                # non WAMP-handled opcides
                if frame.opcode == frame.OPCODE_PING:
                    # Opcode 0x9 marks a ping frame. It does not contain wamp
                    # data, so the frame is not returned.
                    # Still it must be handled or the server will close the
                    # connection.
                    async_adapter.spawn(self.handle_ping(ping_frame=frame))
                    received_bytes = bytearray()
                    continue
                if frame.opcode == frame.OPCODE_PONG:
                    self.handle_pong(pong_frame=frame)
                    received_bytes = bytearray()
                    continue
                if frame.opcode == frame.OPCODE_BINARY:
                    break
                if frame.opcode == frame.OPCODE_CLOSE:
                    self.handle_close(close_frame=frame)

                break

        if frame is None:
            raise NoFrameReturnedError()

        if frame.opcode not in (
            frame.OPCODE_TEXT, frame.OPCODE_CLOSE, frame.OPCODE_BINARY
        ):
            raise WampProtocolError(
                f"Non Text frame returned: {frame.opcode}"
            )

        logger.info(f"returning {frame.opcode}")
        return frame

    def _connect(self):
        if self.ipv == 4:
            _socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

            try:
                _socket.connect((self.host.encode(), self.port))
            except socket_error as exc:
                if exc.errno == 61:
                    logger.error(
                        'unable to connect to %s:%s (IPV%s)',
                        self.host, self.port, self.ipv
                    )
<commit_msg>arg 2 should be a string<commit_after># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import logging
import sched
import socket
import ssl
import uuid
try:
    from base64 import encodestring as encodebytes
except ImportError:
    # who the hell OKed this backwards incompatible change??
    from base64 import encodebytes
from socket import error as socket_error
from time import time

from wampy.backends import async_adapter
from wampy.backends.errors import WampyTimeOut
from wampy.config.defaults import heartbeat, heartbeat_timeout
from wampy.constants import (
    WEBSOCKET_SUBPROTOCOLS, WEBSOCKET_VERSION,
)
from wampy.errors import (
    IncompleteFrameError, NoFrameReturnedError, WampProtocolError, WampyError,
)
from wampy.interfaces import Transport
from wampy.mixins import ParseUrlMixin
from wampy.serializers import json_serialize

from . frames import FrameFactory, Ping, Pong, Text

logger = logging.getLogger(__name__)


class WebSocket(Transport, ParseUrlMixin):

    def __init__(self, server_url, ipv=4):

        self.url = server_url
        self.ipv = ipv

        self.host = None
        self.port = None
        self.resource = None

        self.parse_url()
        self.websocket_location = self.resource
        self.key = encodebytes(uuid.uuid4().bytes).decode('utf-8').strip()
        self.socket = None
        self.connected = False

        self._first_pinged_at = None
        self._pinged_at = None
        self._pong_pointer = None

        self.pongs = async_adapter.queue()
        self.missed_pongs = 0
        self.is_pinging = False

    def connect(self, upgrade=True):
        # TCP connection
        self._connect()
        self._handshake(upgrade=upgrade)

        if heartbeat > 0:
            self.start_pinging()
        return self

    def disconnect(self):
        logger.warning("disconnecting from %s", self.url)
        if self.socket:
            try:
                self.socket.shutdown(socket.SHUT_RDWR)
            except socket.error:
                pass

            self.socket.close()

    def send(self, message):
        frame = Text(payload=json_serialize(message))
        websocket_message = frame.frame
        self._send_raw(websocket_message)

    def _send_raw(self, websocket_message):
        logger.debug('send raw: %s', websocket_message)
        self.socket.sendall(websocket_message)

    def receive(self, bufsize=1):
        frame = None
        received_bytes = bytearray()
        conn = self.socket

        while True:

            try:
                bytes_ = conn.recv(bufsize)
            except socket.timeout as e:
                message = str(e)
                raise ConnectionError('timeout: "{}"'.format(message))
            except OSError:
                logger.info("socket connection lost? Bad File Descriptor?")
                break
            except Exception as exc:
                raise ConnectionError('Connection lost: "{}"'.format(exc))
            if not bytes_:
                break

            received_bytes.extend(bytes_)

            try:
                frame = FrameFactory.from_bytes(received_bytes)
            except IncompleteFrameError as exc:
                bufsize = exc.required_bytes
            else:
                # non WAMP-handled opcides
                if frame.opcode == frame.OPCODE_PING:
                    # Opcode 0x9 marks a ping frame. It does not contain wamp
                    # data, so the frame is not returned.
                    # Still it must be handled or the server will close the
                    # connection.
                    async_adapter.spawn(self.handle_ping(ping_frame=frame))
                    received_bytes = bytearray()
                    continue
                if frame.opcode == frame.OPCODE_PONG:
                    self.handle_pong(pong_frame=frame)
                    received_bytes = bytearray()
                    continue
                if frame.opcode == frame.OPCODE_BINARY:
                    break
                if frame.opcode == frame.OPCODE_CLOSE:
                    self.handle_close(close_frame=frame)

                break

        if frame is None:
            raise NoFrameReturnedError()

        if frame.opcode not in (
            frame.OPCODE_TEXT, frame.OPCODE_CLOSE, frame.OPCODE_BINARY
        ):
            raise WampProtocolError(
                f"Non Text frame returned: {frame.opcode}"
            )

        logger.info(f"returning {frame.opcode}")
        return frame

    def _connect(self):
        if self.ipv == 4:
            _socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

            try:
                _socket.connect((self.host, self.port))
            except socket_error as exc:
                if exc.errno == 61:
                    logger.error(
                        'unable to connect to %s:%s (IPV%s)',
                        self.host, self.port, self.ipv
                    )

{'fraction_non_alphanumeric': 0.07518133699274653, 'fraction_numerical': 0.0038227798470888063, 'mean_word_length': 1.8365304420350292, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2221546', 'n_tokens_mistral': 2843, 'n_tokens_neox': 2687, 'n_words': 737}