def _stream_put_object(self, bucket_name, object_name,
                           data, content_size,
                           metadata=None, sse=None,
                           progress=None, part_size=MIN_PART_SIZE):
        """
        Streaming multipart upload operation.

        :param bucket_name: Bucket name of the multipart upload.
        :param object_name: Object name of the multipart upload.
        :param content_size: Total size of the content to be uploaded.
        :param content_type: Content type of of the multipart upload.
           Defaults to 'application/octet-stream'.
        :param metadata: Any additional metadata to be uploaded along
           with your object.
        :param progress: A progress object
        :param part_size: Multipart part size
        """
        is_valid_bucket_name(bucket_name)
        is_non_empty_string(object_name)
        if not callable(getattr(data, 'read')):
            raise ValueError(
                'Invalid input data does not implement a callable read() method')

        # get upload id.
        upload_id = self._new_multipart_upload(bucket_name, object_name,
                                               metadata, sse)

        # Initialize variables
        total_uploaded = 0
        uploaded_parts = {}

        # Calculate optimal part info.
        total_parts_count, part_size, last_part_size = optimal_part_info(
            content_size, part_size)

        # Instantiate a thread pool with 3 worker threads
        pool = ThreadPool(_PARALLEL_UPLOADERS)
        pool.start_parallel()

        # Generate new parts and upload <= current_part_size until
        # part_number reaches total_parts_count calculated for the
        # given size. Additionally part_manager() also provides
        # md5digest and sha256digest for the partitioned data.
        for part_number in range(1, total_parts_count + 1):
            current_part_size = (part_size if part_number < total_parts_count
                                 else last_part_size)

            part_data = read_full(data, current_part_size)
            pool.add_task(self._upload_part_routine, (bucket_name, object_name, upload_id,
                                                      part_number, part_data, sse, progress))

        try:
            upload_result = pool.result()
        except:
            # Any exception that occurs sends an abort on the
            # on-going multipart operation.
            self._remove_incomplete_upload(bucket_name,
                                           object_name,
                                           upload_id)
            raise

        # Update uploaded_parts with the part uploads result
        # and check total uploaded data.
        while not upload_result.empty():
            part_number, etag, total_read = upload_result.get()
            uploaded_parts[part_number] = UploadPart(bucket_name,
                                                     object_name,
                                                     upload_id,
                                                     part_number,
                                                     etag,
                                                     None,
                                                     total_read)

            total_uploaded += total_read

        if total_uploaded != content_size:
            msg = 'Data uploaded {0} is not equal input size ' \
                  '{1}'.format(total_uploaded, content_size)
            # cleanup incomplete upload upon incorrect upload
            # automatically
            self._remove_incomplete_upload(bucket_name,
                                           object_name,
                                           upload_id)
            raise InvalidSizeError(msg)

        # Complete all multipart transactions if possible.
        try:
            mpart_result = self._complete_multipart_upload(bucket_name,
                                                           object_name,
                                                           upload_id,
                                                           uploaded_parts)
        except:
            # Any exception that occurs sends an abort on the
            # on-going multipart operation.
            self._remove_incomplete_upload(bucket_name,
                                           object_name,
                                           upload_id)
            raise

        # Return etag here.
        return mpart_result.etag
Streaming multipart upload operation.

        :param bucket_name: Bucket name of the multipart upload.
        :param object_name: Object name of the multipart upload.
        :param content_size: Total size of the content to be uploaded.
        :param content_type: Content type of of the multipart upload.
           Defaults to 'application/octet-stream'.
        :param metadata: Any additional metadata to be uploaded along
           with your object.
        :param progress: A progress object
        :param part_size: Multipart part size