    def _internal_in_filter(self, DataQueueClass):
        '''
        url, image -> url, image, filter_image

        input_schema: image
        output_schema: filter_image
        filter_columns: image
        '''

        input_que = DataQueueClass([('url', ColumnType.SCALAR), ('image', ColumnType.QUEUE)])
        output_que = DataQueueClass([('url', ColumnType.SCALAR), ('image', ColumnType.QUEUE), ('filter_image', 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()


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

        def filter_process():
            while True:
                ret = input_que.get()
                if ret is None:
                    break
                i = int(ret[1][len('image'):])
                if i % 2 == 0:
                    output_que.put_dict({'url': ret[0], 'image': ret[1]})
                    continue

                url, image = ret
                filter_image = 'filter_image' + str(i)
                output_que.put((url, image, filter_image))

        t2 = threading.Thread(target=filter_process)
        t2.start()

        t1.join()
        t2.join()
        self.assertEqual(output_que.size, size / 2)
        ret = []
        while output_que.size > 0:
            ret.append(output_que.get())

        output_que.seal()
        self.assertEqual(output_que.size, size / 2)
        while output_que.size > 0:
            ret.append(output_que.get())
url, image -> url, image, filter_image

input_schema: image
output_schema: filter_image
filter_columns: image
tests/unittests/runtime/test_data_queue.py