Rajat Bhardwaj
Rajat Bhardwaj

Reputation: 169

Implementing long polling with python multiprocessing but doesn't work

Here's what I am doing:

def _wait_for_long_poll_resp(param_one, q: Queue):
    print("long poll called", flush=True)
    sys.stdout.flush()
    request_url = <some_url>
    auth = <auth>
    while True:
        print(f'I am Calling {request_url}', flush=True)
        sys.stdout.flush()
        resp = requests.get(request_url, headers=HEADERS, auth=auth)
        print(resp.status_code, flush=True)
        sys.stdout.flush()
        print(resp.content, flush=True)
        sys.stdout.flush()
        if resp.status_code == 200:
            print("Response code is 200", flush=True)
            print(resp.content, flush=True)
            long_poll_resp = LongPollResponse(
                status_code=resp.status_code,
                content=resp.content
            )
        else:
            print("Response code is not 200", flush=True)
            print(resp.status_code, flush=True)
            long_poll_resp = LongPollResponse(status_code=resp.status_code)

        q.put(long_poll_resp)


@contextmanager
def long_pool_listener_for_charger(param_one):
    q = Queue()
    p = Process(target=_wait_for_long_poll_resp, args=[charge_point, q])
    try:
        print("In try block")
        p.start()
        sleep(2.0)
        yield q
    except Exception as e:
        raise Exception("Error occurred: ", e)
    finally:
        p.join()


@contextmanager
def long_pool_listener(param_one):
    with long_pool_listener_for_charger(param_one) as listener:
        yield listener

And then I am writing a test for this using pytest:

def test_remote_reset():
    param_one = "something"
    with long_pool_listener(param_one) as q:  # type: Queue
        sleep(3)
        request_url = <some_url>
        headers = <some_headers>
        auth = <auth>
        print(f'Calling {request_url}')
        resp = requests.post(request_url, headers=headers, auth=auth)
        # Give it some time
        sleep(3)
        resp = q.get(timeout=15)
        # Some assertions

Is this code correct? Because it doesn't seem to even enter the function, _wait_for_long_poll_resp. I know this because it's not even printing the debug prints I have written here. Also wanted to know if it's pythonic enough.

Upvotes: 0

Views: 93

Answers (0)

Related Questions