@loop(fin='stop', exit_exceptions=(socket.error, ))
    def read(self):
        # designed to time out approximately "push_job_interval" after the user
        # last recieved a job. Some miners will consider the mining server dead
        # if they don't recieve something at least once a minute, regardless of
        # whether a new job is _needed_. This aims to send a job _only_ as
        # often as needed
        line = with_timeout(time.time() - self.last_job_push + self.config['push_job_interval'] - self.time_seed,
                            self.fp.readline,
                            timeout_value='timeout')

        if line == 'timeout':
            t = time.time()
            # Set idle status if they haven't submitted in the perscribed time
            if not self.idle and (t - self.last_share_submit) > self.config['idle_worker_threshold']:
                self.idle = True
                self.server.idle_clients += 1

            # Disconnect if we havne't heard from them in a while
            if (t - self.last_share_submit) > self.config['idle_worker_disconnect_threshold']:
                self.logger.info("Disconnecting worker {}.{} at ip {} for inactivity"
                                 .format(self.address, self.worker, self.peer_name[0]))
                self.stop()

            # push a new job if
            if (self.authenticated is True and  # don't send to non-authed
                # force send if we need to push a new difficulty
                (self.next_diff != self.difficulty or
                    # send if we're past the push interval
                    t > (self.last_job_push +
                         self.config['push_job_interval'] -
                         self.time_seed))):

                # Since they might not be submitting jobs due to low hashrate,
                # check vardiff on timeout in addition to on mining submit
                if self.config['vardiff']['enabled'] is True:
                    # If recalc didn't need to adjust then we need to push
                    # because we're timed out
                    if not self.recalc_vardiff():
                        self.push_job(timeout=True)
                else:
                    self.push_job(timeout=True)

            # Continue loop, we just sent new job after timeout
            return

        line = line.strip()

        # Reading from a defunct connection yeilds an EOF character which gets
        # stripped off
        if not line:
            raise LoopExit("Closed file descriptor encountered")

        try:
            data = json.loads(line)
        except __HOLE__:
            self.logger.warn("Data {}.. not JSON".format(line[:15]))
            self.send_error()
            self._incr('unk_err')
            return

        # handle malformed data
        data.setdefault('id', 1)
        data.setdefault('params', [])

        if __debug__:
            self.logger.debug("Data {} recieved on client {}".format(data, self._id))

        # run a different function depending on the action requested from
        # user
        if 'method' not in data:
            self.logger.warn("Empty action in JSON {}".format(self.peer_name[0]))
            self._incr('unk_err')
            self.send_error(id_val=data['id'])
            return

        meth = data['method'].lower()
        if meth == 'mining.subscribe':
            if self.subscribed is True:
                self.send_error(id_val=data['id'])
                return

            try:
                self.client_type = data['params'][0]
            except IndexError:
                pass
            ret = {
                'result': (
                    (
                        # These values aren't used for anything, although
                        # perhaps they should be
                        ("mining.set_difficulty", self._id),
                        ("mining.notify", self._id)
                    ),
                    self._id,
                    self.manager.config['extranonce_size']
                ),
                'error': None,
                'id': data['id']
            }
            self.subscribed = True
            if __debug__:
                self.logger.debug("Sending subscribe response: {}"
                                  .format(pformat(ret)))
            self.write_queue.put(json.dumps(ret) + "\n")

        elif meth == "mining.authorize":
            if self.subscribed is False:
                self._incr('not_subbed_err')
                self.send_error(25, id_val=data['id'])
                return

            if self.authenticated is True:
                self._incr('not_authed_err')
                self.send_error(24, id_val=data['id'])
                return

            try:
                password = data['params'][1]
                username = data['params'][0]
                # allow the user to use the password field as an argument field
                try:
                    args = password_arg_parser.parse_args(password.split())
                except ArgumentParserError:
                    # Ignore malformed parser data
                    pass
                else:
                    if args.diff:
                        diff = max(self.config['minimum_manual_diff'], args.diff)
                        self.difficulty = diff
                        self.next_diff = diff
            except IndexError:
                password = ""
                username = ""

            self.manager.log_event(
                "{name}.auth:1|c".format(name=self.manager.config['procname']))

            self.logger.info("Authentication request from {} for username {}"
                             .format(self.peer_name[0], username))
            user_worker = self.convert_username(username)

            # unpack into state dictionary
            self.address, self.worker = user_worker
            self.authenticated = True
            self.server.set_user(self)

            # notify of success authing and send him current diff and latest
            # job
            self.send_success(data['id'])
            self.push_difficulty()
            self.push_job()

        elif meth == "mining.submit":
            if self.authenticated is False:
                self._incr('not_authed_err')
                self.send_error(24, id_val=data['id'])
                return

            t = time.time()
            diff, typ = self.submit_job(data, t)
            # Log the share to our stat counters
            key = ""
            if typ > 0:
                key += "reject_"
            key += StratumClient.share_type_strings[typ] + "_share"
            if typ == 0:
                # Increment valid shares to calculate hashrate
                self._incr(key + "_n1", diff)
            self.manager.log_event(
                "{name}.{type}:1|c\n"
                "{name}.{type}_n1:{diff}|c\n"
                "{name}.submit_time:{t}|ms"
                .format(name=self.manager.config['procname'], type=key,
                        diff=diff, t=(time.time() - t) * 1000))

            # don't recalc their diff more often than interval
            if (self.config['vardiff']['enabled'] is True and
                    (t - self.last_diff_adj) > self.config['vardiff']['interval']):
                self.recalc_vardiff()

        elif meth == "mining.get_transactions":
            self.send_error(id_val=data['id'])
        elif meth == "mining.extranonce.subscribe":
            self.send_success(id_val=data['id'])

        else:
            self.logger.info("Unkown action {} for command {}"
                             .format(data['method'][:20], self.peer_name[0]))
            self._incr('unk_err')
            self.send_error(id_val=data['id'])
dataset/ETHPy150Open simplecrypto/powerpool/powerpool/stratum_server.py/StratumClient.read