Thread Method, running the connection for each endpoint.
        :param endpoint:
        :return:
550041
def _subscription_thread(self, endpoint):
        
        try:
            conn = create_connection(self.addr + endpoint, timeout=5)
        except WebSocketTimeoutException:
            self.restart_q.put(endpoint)
            return

        while self.threads_running[endpoint]:
            try:
                msg = conn.recv()
            except WebSocketTimeoutException:
                self._controller_q.put(endpoint)

            log.debug("%s, %s", endpoint, msg)
            ep, pair = endpoint.split('/')
            log.debug("_subscription_thread(): Putting data on q..")
            try:
                self.data_q.put((ep, pair, msg, time.time()), timeout=1)
            except TimeoutError:
                continue
            finally:
                log.debug("_subscription_thread(): Data Processed, looping back..")
        conn.close()
        log.debug("_subscription_thread(): Thread Loop Ended.")
927086
def connectCached(self, endpoint, protocolFactory,
                      extraWork=lambda x: x,
                      extraHash=None):
        
        key = endpoint, extraHash
        D = Deferred()
        if key in self.cachedConnections:
            D.callback(self.cachedConnections[key])
        elif key in self.inProgress:
            self.inProgress[key].append(D)
        else:
            self.inProgress[key] = [D]
            endpoint.connect(
                _CachingClientFactory(
                    self, key, protocolFactory,
                    extraWork))
        return D
1235409
def bind(endpoint)
      rc = ffi_delegate.bind("%s", :string, endpoint)
      raise_zmq_err("unable to bind to %p" % endpoint) if rc == -1
      @last_tcp_port = rc if rc > 0
    end
1020271
async def run_async(self, endpoint: str):
        
        self._connect(endpoint)

        # spawn an initial listen task
        listen_task = asyncio.ensure_future(self._socket.recv_multipart())
        task_list = [listen_task]

        while True:
            dones, pendings = await asyncio.wait(task_list, return_when=asyncio.FIRST_COMPLETED)

            # grab one "done" task to handle
            task_list, done_list = list(pendings), list(dones)
            done = done_list.pop()
            task_list += done_list

            if done == listen_task:
                try:
                    # empty_frame may either be:
                    # 1. a single null frame if the client is a REQ socket
                    # 2. an empty list (ie. no frames) if the client is a DEALER socket
                    identity, *empty_frame, msg = done.result()
                    request = from_msgpack(msg)

                    # spawn a processing task
                    task_list.append(asyncio.ensure_future(
                        self._process_request(identity, empty_frame, request)))
                except Exception as e:
                    if self.serialize_exceptions:
                        _log.exception('Exception thrown in Server run loop during request '
                                       'reception: {}'.format(str(e)))
                    else:
                        raise e
                finally:
                    # spawn a new listen task
                    listen_task = asyncio.ensure_future(self._socket.recv_multipart())
                    task_list.append(listen_task)
            else:
                # if there's been an exception during processing, consider reraising it
                try:
                    done.result()
                except Exception as e:
                    if self.serialize_exceptions:
                        _log.exception('Exception thrown in Server run loop during request '
                                       'dispatch: {}'.format(str(e)))
                    else:
                        raise e
286409
def endpoint_class(collection):
    
    with app.app_context():
        try:
            cls = current_app.class_references[collection]
        except KeyError:
            raise InvalidAPIUsage(404)
        return cls
936231
def api_server(connection, server_class):
    

    # run api server on connection
    return server_class(
        link=xbahn.connection.link.Link(
            # use the connection to receive messages
            receive=connection,
            # use the connection to respond to received messages
            respond=connection
        )
    )
603779
def run(self):
        
        target = getattr(self, '_Thread__target', getattr(self, '_target', None))
        args = getattr(self, '_Thread__args', getattr(self, '_args', None))
        kwargs = getattr(self, '_Thread__kwargs', getattr(self, '_kwargs', None))
        if target is not None:
            self._return = target(*args, **kwargs)

        return None
647787
def post(self, endpoint, **kwargs):
        
        return self._request(requests.post, endpoint, **kwargs)
218280
private ImmutableList<EndPoint> getEndPoints() {
        ImmutableList<EndPoint> endPoints = _endPoints;
        if (endPoints == null) {
            boolean locked = false;
            try {
                long timeout = System.currentTimeMillis() + 10;

                locked = _lock.tryLock(10, TimeUnit.MILLISECONDS);
                if (locked) {
                    endPoints = _endPoints;
                    long waitTime = timeout - System.currentTimeMillis();
                    if (endPoints == null && waitTime > 0) {
                        _endpointsAvailable.await(waitTime, TimeUnit.MILLISECONDS);
                        endPoints = _endPoints;
                    }
                }
            } catch (InterruptedException e) {
                // ignore
            } finally {
                if (locked) {
                    _lock.unlock();
                }
            }

            if (endPoints == null) {
                // Took too long waiting for the background to re-populate endpoints, just fetch from the source
                endPoints = getEndPointsFromHostDiscovery();
            }
        }
        
        return endPoints;
    }
1059851
def _connect_to_socket(self, context: zmq.Context, endpoint: str):
        
        socket = context.socket(zmq.DEALER)
        socket.connect(endpoint)
        socket.setsockopt(zmq.LINGER, 0)
        _log.debug("Client connected to endpoint %s", self.endpoint)
        return socket
368047
def _thread_body(self, request, callback):
        
        self.protocol.send_message(request)
        while not self.protocol.stopped.isSet():
            response = self.queue.get(block=True)
            callback(response)
437508
def request(self, *args, **kwargs):
        
        func = partial(super().request, *args, **kwargs)
        return self.loop.run_in_executor(self.thread_pool, func)
551027
def join(self, *args, **kwargs):
        
        super(ThreadReturn, self).join(*args, **kwargs)
        return self._return
150279
def from_tasks(cls, tasks):
        

        endpoints = [
            [
                MarathonEndpoint(task.app_id, task.service_ports[
                                 port_index], task.host, task.id, port)
                for port_index, port in enumerate(task.ports)
            ]
            for task in tasks
        ]
        # Flatten result
        return [item for sublist in endpoints for item in sublist]
1216558
def _threaded_callback(callback, *args):
    
    t = Thread(target=callback, args=args)
    t.daemon = True
    t.start()
44143
public void run()
        {
            outer:
            while (run || !queue.isEmpty())
            {
                List<ByteBuffer> bindVariables;
                try
                {
                    bindVariables = queue.take();
                }
                catch (InterruptedException e)
                {
                    // re-check loop condition after interrupt
                    continue;
                }

                Iterator<InetAddress> iter = endpoints.iterator();
                while (true)
                {
                    // send the mutation to the last-used endpoint.  first time through, this will NPE harmlessly.
                    try
                    {
                        int i = 0;
                        int itemId = preparedStatement(client);
                        while (bindVariables != null)
                        {
                            client.execute_prepared_cql3_query(itemId, bindVariables, ConsistencyLevel.ONE);
                            i++;
                            
                            if (i >= batchThreshold)
                                break;
                            
                            bindVariables = queue.poll();
                        }
                        
                        break;
                    }
                    catch (Exception e)
                    {
                        closeInternal();
                        if (!iter.hasNext())
                        {
                            lastException = new IOException(e);
                            break outer;
                        }
                    }

                    // attempt to connect to a different endpoint
                    try
                    {
                        InetAddress address = iter.next();
                        String host = address.getHostName();
                        int port = ConfigHelper.getOutputRpcPort(conf);
                        client = CqlOutputFormat.createAuthenticatedClient(host, port, conf);
                    }
                    catch (Exception e)
                    {
                        closeInternal();
                        // TException means something unexpected went wrong to that endpoint, so
                        // we should try again to another.  Other exceptions (auth or invalid request) are fatal.
                        if ((!(e instanceof TException)) || !iter.hasNext())
                        {
                            lastException = new IOException(e);
                            break outer;
                        }
                    }
                }
            }

            // close all our connections once we are done.
            closeInternal();
        }
650003
def post(self, endpoint: str, **kwargs) -> dict:
        """"""

        return self._request('POST', endpoint, **kwargs)