    def process_step(self):
        """
        Process each window data.
        """
        self._time_profiler.record(self.uid, Event.queue_in)
        in_buffer = self._get_buffer()
        if in_buffer is None:
            return

        if all(map(lambda x: len(x) == 0, in_buffer.values())):
            self._set_finished()
            return

        process_data = [in_buffer.get(key) for key in self._node_repr.inputs]
        self._time_profiler.record(self.uid, Event.process_in)
        succ, outputs, msg = self._call(process_data)
        self._time_profiler.record(self.uid, Event.process_out)
        assert succ, msg

        size = len(self._node_repr.outputs)
        if size > 1:
            output_map = dict((self._node_repr.outputs[i], outputs[i])
                              for i in range(size))
        elif size == 1:
            output_map = {}
            output_map[self._node_repr.outputs[0]] = outputs
        else:
            output_map = {}

        self._time_profiler.record(self.uid, Event.queue_out)
        if not self.data_to_next(output_map):
            return
        self._set_finished()
Process each window data.
towhee/runtime/nodes/_window_all.py