    def _internal_in_window(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(size):
                input_que.put(('http://towhee.io', 'image' + str(i)))
                time.sleep(0.01)
            input_que.seal()

        def new_container():
            items = []
            for _ in range(output_que.col_size):
                items.append([])
            return items

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

        batch = 3
        i = 0

        items = new_container()
        while True:
            ret = input_que.get()

            if ret is None:
                if not items[0]:
                    break
                else:
                    vecs = ['vec', 'vec']
                    items[-1] = vecs
                    output_que.batch_put(items)
                    items = new_container()
                    break

            for i in range(len(ret)):
                items[i].append(ret[i])

            if len(items[0]) == batch:
                vecs = ['vec', 'vec']
                items[-1] = vecs
                output_que.batch_put(items)
                items = new_container()

        output_que.seal()
        self.assertEqual(output_que.size, size)
        i = 0
        vec_size = size // batch
        if size % batch != 0:
            vec_size += 1
        vec_size = vec_size * 2
        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))
            if i < vec_size:
                self.assertEqual(ret[2], 'vec')
            else:
                self.assertIsInstance(ret[2], Empty)
            i += 1
        t.join()
url, image -> url, image, vec

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