def test_send_news(support):
    address = nw0.core.address()
    topic = uuid.uuid4().hex
    data = uuid.uuid4().hex
    reply_queue = queue.Queue()
    
    support.queue.put(("send_news_to", [address, topic, reply_queue]))
    while True:
        nw0.send_news_to(address, topic, None)
        try:
            in_topic, in_data = reply_queue.get_nowait()
        except queue.Empty:
            time.sleep(0.1)
        else:
            break

    nw0.send_news_to(address, topic, data)
    while in_data is None:
        in_topic, in_data = reply_queue.get()
    
    assert in_topic, in_data == (topic, data)

#
# wait_for_news_from
# 

def test_wait_for_news(support):
    address = nw0.core.address()
    topic = uuid.uuid4().hex
    data = uuid.uuid4().hex
    sync_queue = queue.Queue()
    
    support.queue.put(("wait_for_news_from", [address, topic, data, sync_queue]))
    in_topic, in_data = nw0.wait_for_news_from(address, topic, wait_for_s=5)
    sync_queue.put(True)
    while in_data is None:
        in_topic, in_data = nw0.wait_for_news_from(address, topic, wait_for_s=5)
    assert (topic, data) == (in_topic, in_data)

#
# send to multiple addresses
# For now, this is disallowed as the semantics aren't
# intuitive. (It does a round-robin selection which is useful
# for things like load-scheduling but not for broadcasting).
# 

def test_arg_passing(self):
        #Make sure that parameter passing works.
        def arg_tester(queue, arg1=False, arg2=False):
            """Use to test _thread.start_new_thread() passes args properly."""
            queue.put((arg1, arg2))

        testing_queue = queue.Queue(1)
        _thread.start_new_thread(arg_tester, (testing_queue, True, True))
        result = testing_queue.get()
        self.assertTrue(result[0] and result[1],
                        "Argument passing for thread creation using tuple failed")
        _thread.start_new_thread(arg_tester, tuple(), {'queue':testing_queue,
                                                       'arg1':True, 'arg2':True})
        result = testing_queue.get()
        self.assertTrue(result[0] and result[1],
                        "Argument passing for thread creation using kwargs failed")
        _thread.start_new_thread(arg_tester, (testing_queue, True), {'arg2':True})
        result = testing_queue.get()
        self.assertTrue(result[0] and result[1],
                        "Argument passing for thread creation using both tuple"
                        " and kwargs failed") 

def test_arg_passing(self):
        #Make sure that parameter passing works.
        def arg_tester(queue, arg1=False, arg2=False):
            """Use to test _thread.start_new_thread() passes args properly."""
            queue.put((arg1, arg2))

        testing_queue = queue.Queue(1)
        _thread.start_new_thread(arg_tester, (testing_queue, True, True))
        result = testing_queue.get()
        self.assertTrue(result[0] and result[1],
                        "Argument passing for thread creation using tuple failed")
        _thread.start_new_thread(arg_tester, tuple(), {'queue':testing_queue,
                                                       'arg1':True, 'arg2':True})
        result = testing_queue.get()
        self.assertTrue(result[0] and result[1],
                        "Argument passing for thread creation using kwargs failed")
        _thread.start_new_thread(arg_tester, (testing_queue, True), {'arg2':True})
        result = testing_queue.get()
        self.assertTrue(result[0] and result[1],
                        "Argument passing for thread creation using both tuple"
                        " and kwargs failed") 

def unpack_observed(annotation_info, queue, obs_file):
    """ Now that transcript has been annotated, unpack values and
        create an observed entry. Send the observed entry to the queue 
        for output to obs_file."""

    obs_ID = observed_counter.increment()
    observed = (obs_ID, annotation_info.gene_ID, annotation_info.transcript_ID, 
                annotation_info.read_ID, annotation_info.dataset,
                annotation_info.start_vertex, annotation_info.end_vertex, 
                annotation_info.start_exon, annotation_info.end_exon,
                annotation_info.start_delta, annotation_info.end_delta, 
                annotation_info.read_length, annotation_info.fraction_As,
                annotation_info.custom_label, annotation_info.allelic_label,
                annotation_info.start_support, annotation_info.end_support)
    msg = (obs_file, "\t".join([str(x) for x in observed]))
    queue.put(msg)

    return 

def grab(cam, queue, width, height, fps):
    global running
    capture = cv2.VideoCapture(cam)
    capture.set(cv2.CAP_PROP_FRAME_WIDTH, width)
    capture.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
    capture.set(cv2.CAP_PROP_FPS, fps)

    while(running):
        frame = {}
        capture.grab()
        retval, img = capture.retrieve(0)
        frame["img"] = img
        frame["1"] = config["1"]
        frame["2"] = config["2"]

        blur = get_blur(img, 0.05)
        frame["blur"] = blur

        if queue.qsize() < 10:
            queue.put(frame)
        else:
            print(queue.qsize()) 

def grab(cam, queue, width, height, fps):
    global running
    capture = cv2.VideoCapture(cam)
    capture.set(cv2.CAP_PROP_FRAME_WIDTH, width)
    capture.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
    capture.set(cv2.CAP_PROP_FPS, fps)

    while(running):
        frame = {}
        capture.grab()
        retval, img = capture.retrieve(0)
        frame["img"] = img
        frame["1"] = config["1"]
        frame["2"] = config["2"]

        blur = get_blur(img, 0.05)
        frame["blur"] = blur

        if queue.qsize() < 10:
            queue.put(frame)
        else:
            print(queue.qsize()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits
    curr_thread = threading.current_thread()
    THREADS.add(curr_thread.ident)

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits
    curr_thread = threading.current_thread()
    THREADS.add(curr_thread.ident)

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish_sync(queue):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits
    curr_thread = threading.current_thread()
    THREADS.add(curr_thread.ident)

    while True:
        msg_id = str(uuid.uuid4())
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=msg_id, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {msg}")
        # simulate randomness of publishing messages
        time.sleep(random.random()) 

def publish(queue, n):
    """Simulates an external publisher of messages.

    Args:
        queue (queue.Queue): Queue to publish messages to.
        n (int): Number of messages to publish.
    """
    choices = string.ascii_lowercase + string.digits
    for x in range(1, n + 1):
        host_id = "".join(random.choices(choices, k=4))
        instance_name = f"cattle-{host_id}"
        msg = PubSubMessage(message_id=x, instance_name=instance_name)
        # publish an item
        queue.put(msg)
        logging.info(f"Published {x} of {n} messages")

    # indicate the publisher is done
    queue.put(None) 

def _StartProcess(self):
        print("Command: {}".format(self.arguments))
        self.process = Popen(self.arguments, stderr=STDOUT, stdout=PIPE,
                             env=os.environ, bufsize=1, close_fds=isPosix())
        self.pid = self.process.pid

        def enqueue_output(out, queue):
            for line in iter(out.readline, ""):
                queue.put(line)
            out.close()

        self.terminal_queue = Queue.Queue()
        self.terminal_producer = threading.Thread(target=enqueue_output, args=(self.process.stdout, self.terminal_queue))
        self.terminal_consumer = threading.Thread(target=self._grab_sanitizer_trace)
        self.terminal_producer.setDaemon(True)
        self.terminal_consumer.setDaemon(True)
        self.terminal_producer.start()
        self.terminal_consumer.start() 

def on_sub_mock(mocker):

    mgr = multiprocess.Manager()
    q = mgr.Queue()

    def on_subscribe(self, msg, params, websocket):
        new_params = copy.deepcopy(params)
        new_params.update({'context': msg.get('context', {})})
        q.put(self)
        return new_params

    on_sub_mock = {
        'on_subscribe':
        PickableMock(side_effect=promisify(on_subscribe), name='on_subscribe')
    }

    return on_sub_mock, q 

def producer(pool, queue, submap_ft, refmap_ft, fname, particles,
             sx, sy, s, a, apix, coefs_method, r, nr, fftthreads=1, crop=None, pfac=2):
    log = logging.getLogger('root')
    log.debug("Producing %s" % fname)
    zreader = mrc.ZSliceReader(particles[star.UCSF.IMAGE_ORIGINAL_PATH].iloc[0])
    for i, ptcl in particles.iterrows():
        log.debug("Produce %d@%s" % (ptcl[star.UCSF.IMAGE_ORIGINAL_INDEX], ptcl[star.UCSF.IMAGE_ORIGINAL_PATH]))
        # p1r = mrc.read_imgs(stack[i], idx[i] - 1, compat="relion")
        p1r = zreader.read(ptcl[star.UCSF.IMAGE_ORIGINAL_INDEX])
        log.debug("Apply")
        ri = pool.apply_async(
            subtract_outer,
            (p1r, ptcl, submap_ft, refmap_ft, sx, sy, s, a, apix, coefs_method, r, nr),
            {"fftthreads": fftthreads, "crop": crop, "pfac": pfac})
        log.debug("Put")
        queue.put((ptcl[star.UCSF.IMAGE_INDEX], ri), block=True)
        log.debug("Queue for %s is size %d" % (ptcl[star.UCSF.IMAGE_ORIGINAL_PATH], queue.qsize()))
    zreader.close()
    log.debug("Put poison pill")
    queue.put((-1, None), block=True) 

def mock_queue(monkeypatch):

    class MockQueue:
        items = []

        def get(self, timeout=None):
            try:
                return self.items.pop()
            except IndexError:
                if timeout:
                    raise queue.Empty()
                raise

        def put(self, item):
            self.items.append(item)

    mockqueue = MockQueue()

    monkeypatch.setattr('queue.Queue', lambda: mockqueue)
    return mockqueue 

def mock_queue(monkeypatch):

    class MockQueue:
        items = []

        def get(self, timeout=None):
            try:
                return self.items.pop()
            except IndexError:
                if timeout:
                    raise queue.Empty()
                raise

        def put(self, item):
            self.items.append(item)

    mockqueue = MockQueue()

    monkeypatch.setattr('queue.Queue', lambda: mockqueue)
    return mockqueue 

def source_queue(self, procs, queues):
        while True:
            for index, proc in enumerate(procs):
                try:
                    queue = queues[index]
                    h, w, c = self.shape[index]
                    frames_size = w * h * c
                    nv12_data = proc.stdout.read(frames_size)
                    if len(nv12_data) != frames_size:
                        # logger.error("source_queue read error index %s len %s", index, len(nv12_data))
                        continue
                    frame = np.frombuffer(nv12_data, dtype=np.uint8)
                    img = np.array(frame, dtype=np.uint8).reshape((h, w, c))
                    if DEBUG:
                        queue.put(img)
                    else:
                        queue.put_nowait(img)
                except Exception as e:
                    # logger.error("source_queue queue full index %s", index)
                    pass 

def myPublisher(queue):
  while not queue.full():
    queue.put(1)
    print("{} Appended 1 to queue: {}".format(threading.current_thread(), queue.qsize()))
    time.sleep(1) 

def myThread(queue):
  while True:
    try:
      time.sleep(2)
      raise Exception("Exception Thrown In Child Thread {}".format(threading.current_thread()))
    except:
      queue.put(sys.exc_info()) 

def prefetch_data(system_config, db, queue, sample_data, data_aug):
    ind = 0
    print("start prefetching data...")
    # 用pid[进程id]做随机数种子
    np.random.seed(os.getpid())
    while True:
        try:
            data, ind = sample_data(system_config, db, ind, data_aug=data_aug)
            queue.put(data)
        except Exception as e:
            traceback.print_exc()
            raise e 

def pin_memory(data_queue, pinned_data_queue, sema):
    while True:
        data = data_queue.get()

        data["xs"] = [_pin_memory(x) for x in data["xs"]]
        data["ys"] = [_pin_memory(y) for y in data["ys"]]

        pinned_data_queue.put(data)

        if sema.acquire(blocking=False):
            return 

def prefetch_data(system_config, db, queue, sample_data, data_aug):
    ind = 0
    print("start prefetching data...")
    # 用pid[进程id]做随机数种子
    np.random.seed(os.getpid())
    while True:
        try:
            data, ind = sample_data(system_config, db, ind, data_aug=data_aug)
            queue.put(data)
        except Exception as e:
            traceback.print_exc()
            raise e 

def pin_memory(data_queue, pinned_data_queue, sema):
    while True:
        data = data_queue.get()

        data["xs"] = [_pin_memory(x) for x in data["xs"]]
        data["ys"] = [_pin_memory(y) for y in data["ys"]]

        pinned_data_queue.put(data)

        if sema.acquire(blocking=False):
            return 

def echoreverse(ip, queue, data):
    # Reverse the data.
    # result = reversed(data)
    # Convert each byte to a character.
    # result = [chr(b) for b in reversed(data)]
    # Turn the list of characters into a string with "" in between each char.
    # result = "".join([chr(b) for b in reversed(data)])
    # Turn the whole thing back into bytes with UTF-8 encoding.
    # result = bytes("".join([chr(b) for b in reversed(data)]), "UTF-8")
    # Put the result back into the queue of data to be sent back:
    queue.put(bytes("".join([chr(b) for b in reversed(data)]), "UTF-8")) 

def echo(ip, queue, data):
    queue.put(data)