91f630f70959098facef4ab4e1a98d1f48f9a586,streaming/python/collector.py,OutputCollector,collect,#OutputCollector#Any#,43

Before Change

    def collect(self, record):
        partitions = self._partition_func.partition(record,
                                                    len(self._channel_ids))
        serialized_message = pickle.dumps(record)
        for partition_index in partitions:
            self._writer.write(self._channel_ids[partition_index],
                               serialized_message)

After Change

        python_buffer = None
        cross_lang_buffer = None
        for partition_index in partitions:
            if self._target_languages[partition_index] == \
                    function.Language.PYTHON:
                // avoid repeated serialization
                if python_buffer is None:
                    python_buffer = self.python_serializer.serialize(record)
                self._writer.write(
                    self._channel_ids[partition_index],
                    serialization._PYTHON_TYPE_ID + python_buffer)
            else:
                // avoid repeated serialization
                if cross_lang_buffer is None:
                    cross_lang_buffer = self.cross_lang_serializer.serialize(
                        record)
                self._writer.write(
                    self._channel_ids[partition_index],
                    serialization._CROSS_LANG_TYPE_ID + cross_lang_buffer)

Instances