    def _internal_in_map(self, DataQueueClass):
        '''
        url, image -> url, image, vec

        input_schema: image
        output_schema: vec
        '''
        input_que = DataQueueClass([('url', ColumnType.SCALAR), ('image', ColumnType.QUEUE)])
        output_que = DataQueueClass([('url', ColumnType.SCALAR), ('image', ColumnType.QUEUE), ('vec', ColumnType.QUEUE)])

        size = 20

        def write():
            for i in range(1, size + 1):
                input_que.put(('http://towhee.io', 'image' + str(i)))
                time.sleep(0.01)
            input_que.seal()

        t = threading.Thread(target=write)
        t.start()

        i = 1
        while True:
            ret = input_que.get()
            if ret is None:
                output_que.seal()
                break
            url, image = ret
            vec = 'vec' + str(i)
            output_que.put((url, image, vec))
            i += 1

        self.assertEqual(output_que.size, size)
        i = 1
        while True:
            ret = output_que.get()
            if ret is None:
                break
            self.assertEqual(ret[0], 'http://towhee.io')
            self.assertEqual(ret[1], 'image' + str(i))
            self.assertEqual(ret[2], 'vec' + str(i))
            i += 1
        t.join()
url, image -> url, image, vec

input_schema: image
output_schema: vec
tests/unittests/runtime/test_data_queue.py