Finalize written chunk
        :param Downloader self: this
        :param blobxfer.models.download.Descriptor dd: download descriptor
        :param blobxfer.models.download.Offsets offsets: offsets
434574
def _finalize_chunk(self, dd, offsets):
        # type: (Downloader, blobxfer.models.download.Descriptor,
        #        blobxfer.models.download.Offsets) -> None
        
        if dd.entity.is_encrypted:
            dd.mark_unchecked_chunk_decrypted(offsets.chunk_num)
        # integrity check data and write to disk (this is called
        # regardless of md5/hmac enablement for resume purposes)
        dd.perform_chunked_integrity_check()
        # remove from disk set and add bytes to counter
        with self._disk_operation_lock:
            self._disk_set.remove(
                blobxfer.operations.download.Downloader.
                create_unique_disk_operation_id(dd, offsets))
            self._download_bytes_sofar += offsets.num_bytes
683000
def _finalize_block_blob(self, sd, metadata, digest):
        # type: (SyncCopy, blobxfer.models.synccopy.Descriptor, dict,
        #        str) -> None
        
        blobxfer.operations.azure.blob.block.put_block_list(
            sd.dst_entity, sd.last_block_num, digest, metadata)
        if blobxfer.util.is_not_empty(sd.dst_entity.replica_targets):
            for ase in sd.dst_entity.replica_targets:
                blobxfer.operations.azure.blob.block.put_block_list(
                    ase, sd.last_block_num, digest, metadata)
176463
def _finalize_block_blob(self, ud, metadata):
        # type: (Uploader, blobxfer.models.upload.Descriptor, dict) -> None
        
        if not ud.entity.is_encrypted and ud.must_compute_md5:
            digest = blobxfer.util.base64_encode_as_string(ud.md5.digest())
        else:
            digest = None
        blobxfer.operations.azure.blob.block.put_block_list(
            ud.entity, ud.last_block_num, digest, metadata)
        if blobxfer.util.is_not_empty(ud.entity.replica_targets):
            for ase in ud.entity.replica_targets:
                blobxfer.operations.azure.blob.block.put_block_list(
                    ase, ud.last_block_num, digest, metadata)
42024
def generate_downloader(headers: Dict[str, str], args: Any, max_per_hour: int=30
                        ) -> Callable[..., None]:
    """"""

    def _downloader(url: str, dest: str) -> None:

        @rate_limited(max_per_hour, args)
        def _rate_limited_download() -> None:

            # Create parent directory of file, and its parents, if they don't exist.
            parent = os.path.dirname(dest)
            if not os.path.exists(parent):
                os.makedirs(parent)

            response = requests.get(url, headers=headers, stream=True)
            LOG.info(f"Downloading from '{url}'.")
            LOG.info(f"Trying to save to '{dest}'.")

            length = response.headers.get("content-length")
            if length is None:
                total_length = 0
            else:
                total_length = int(length)

            expected_size = (total_length / CHUNK_SIZE) + 1
            chunks = response.iter_content(chunk_size=CHUNK_SIZE)

            open(dest, "a", encoding=FORCED_ENCODING).close()
            # per http://stackoverflow.com/a/20943461
            with open(dest, "wb") as stream:
                for chunk in tui.progress.bar(chunks, expected_size=expected_size):
                    if not chunk:
                        return
                    stream.write(chunk)
                    stream.flush()

        _rate_limited_download()

    return _downloader
747689
def download(url, target, headers=None, trackers=()):
    
    if headers is None:
        headers = {}
    headers.setdefault('user-agent', 'requests_download/'+__version__)
    r = requests.get(url, headers=headers, stream=True)
    r.raise_for_status()
    for t in trackers:
        t.on_start(r)


    with open(target, 'wb') as f:
        for chunk in r.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)
                for t in trackers:
                    t.on_chunk(chunk)

    for t in trackers:
        t.on_finish()
746313
def create_unique_disk_operation_id(dd, offsets):
        # type: (blobxfer.models.download.Descriptor,
        #        blobxfer.models.download.Offsets) -> str
        
        return ';'.join(
            (str(dd.final_path), dd.entity._client.primary_endpoint,
             dd.entity.path, str(offsets.range_start))
        )
912437
def download_to_file(self, local_file, headers=None, cookies=None,
                         chunk_size=1024):
        

        if not headers:
            headers = HEADERS

        if not cookies:
            cookies = {}

        req = requests.get(self.url, headers=headers, cookies=cookies,
                           stream=True)
        with open(local_file, "wb") as local_download:
            for chunk in req.iter_content(chunk_size):
                if chunk:
                    local_download.write(chunk)
774856
def next_offsets(self):
        # type: (Descriptor) -> Offsets
        
        resume_bytes = self._resume()
        if resume_bytes is None and not self._allocated:
            self._allocate_disk_space()
        with self._meta_lock:
            if self._offset >= self._ase.size:
                return None, resume_bytes
            if self._offset + self._chunk_size > self._ase.size:
                chunk = self._ase.size - self._offset
            else:
                chunk = self._chunk_size
            # on download, num_bytes must be offset by -1 as the x-ms-range
            # header expects it that way. x -> y bytes means first bits of the
            # (x+1)th byte to the last bits of the (y+1)th byte. for example,
            # 0 -> 511 means byte 1 to byte 512
            num_bytes = chunk - 1
            chunk_num = self._chunk_num
            fd_start = self._offset
            range_start = self._offset
            if self._ase.is_encrypted:
                # ensure start is AES block size aligned
                range_start = range_start - \
                    (range_start % self._AES_BLOCKSIZE) - \
                    self._AES_BLOCKSIZE
                if range_start <= 0:
                    range_start = 0
            range_end = self._offset + num_bytes
            self._offset += chunk
            self._chunk_num += 1
            if self._ase.is_encrypted and self._offset >= self._ase.size:
                unpad = True
            else:
                unpad = False
            return Offsets(
                chunk_num=chunk_num,
                fd_start=fd_start,
                num_bytes=chunk,
                range_start=range_start,
                range_end=range_end,
                unpad=unpad,
            ), resume_bytes
551751
public void downloadTo(Path path, BlobSourceOption... options) {
    try (OutputStream outputStream = Files.newOutputStream(path);
        ReadChannel reader = reader(options)) {
      WritableByteChannel channel = Channels.newChannel(outputStream);
      ByteBuffer bytes = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
      while (reader.read(bytes) > 0) {
        bytes.flip();
        channel.write(bytes);
        bytes.clear();
      }
    } catch (IOException e) {
      throw new StorageException(e);
    }
  }
848610
def _add_to_download_queue(self, lpath, rfile):
        # type: (Downloader, pathlib.Path,
        #        blobxfer.models.azure.StorageEntity) -> None
        
        # prepare remote file for download
        dd = blobxfer.models.download.Descriptor(
            lpath, rfile, self._spec.options, self._general_options,
            self._resume)
        with self._transfer_lock:
            self._transfer_cc[dd.entity.path] = 0
            if dd.entity.is_encrypted:
                self._dd_map[str(dd.final_path)] = dd
        # add download descriptor to queue
        self._transfer_queue.put(dd)
        if self._download_start_time is None:
            with self._transfer_lock:
                if self._download_start_time is None:
                    self._download_start_time = blobxfer.util.datetime_now()
1187277
def _set_final_path_view(self):
        # type: (Descriptor) -> int
        
        # set final path if vectored io stripe
        if self._ase.vectored_io is not None:
            self.final_path = blobxfer.models.download.Descriptor.\
                convert_vectored_io_slice_to_final_path_name(
                    self.final_path, self._ase)
        # generate view
        view, total_size = blobxfer.models.download.Descriptor.generate_view(
            self._ase)
        self.view = view
        return total_size
262381
def download(url, file_handle, chunk_size=1024):
    
    r = requests.get(url, stream=True)
    total_length = r.headers.get('content-length')
    if total_length is None:
        maxval = UnknownLength
    else:
        maxval = int(total_length)
    name = file_handle.name
    with progress_bar(name=name, maxval=maxval) as bar:
        for i, chunk in enumerate(r.iter_content(chunk_size)):
            if total_length:
                bar.update(i * chunk_size)
            file_handle.write(chunk)
453146
def finalise(self, filename, totalChunks, status):
        

        def getChunkIdx(x):
            try:
                return int(x.suffix[1:])
            except ValueError:
                return -1

        def isChunkFile(x):
            return x.is_file() and -1 < getChunkIdx(x) <= totalChunks

        asSingleFile = os.path.join(self._tmpDir, filename)
        if status.lower() == 'true':
            chunks = [(getChunkIdx(file), str(file)) for file in
                      Path(self._tmpDir).glob(filename + '.*') if isChunkFile(file)]
            # TODO if len(chunks) != totalChunks then error
            with open(asSingleFile, 'xb') as wfd:
                for f in [x[1] for x in sorted(chunks, key=lambda tup: tup[0])]:
                    with open(f, 'rb') as fd:
                        logger.info("cat " + f + " with " + asSingleFile)
                        shutil.copyfileobj(fd, wfd, 1024 * 1024 * 10)
        self.cleanupChunks(filename, isChunkFile, status)
845541
def _process_upload_descriptor(self, ud):
        # type: (Uploader, blobxfer.models.upload.Descriptor) -> None
        
        # get upload offsets
        offsets, resume_bytes = ud.next_offsets()
        # add resume bytes to counter
        if resume_bytes is not None:
            with self._transfer_lock:
                self._upload_bytes_total += ud.entity.size
                self._upload_bytes_sofar += resume_bytes
                logger.debug('adding {} sofar {} from {}'.format(
                    resume_bytes, self._upload_bytes_sofar, ud.entity.name))
            del resume_bytes
        # check if all operations completed
        if offsets is None and ud.all_operations_completed:
            # finalize file
            self._finalize_upload(ud)
            # accounting
            with self._upload_lock:
                if ud.entity.is_encrypted:
                    self._ud_map.pop(ud.unique_id)
                self._upload_set.remove(ud.unique_id)
                self._upload_sofar += 1
            return
        # if nothing to upload, re-enqueue for finalization
        if offsets is None:
            self._upload_queue.put(ud)
            return
        # prepare upload
        if offsets.chunk_num == 0:
            self._prepare_upload(ud.entity)
        # prepare replicae targets
        if blobxfer.util.is_not_empty(ud.entity.replica_targets):
            for ase in ud.entity.replica_targets:
                if offsets.chunk_num == 0:
                    self._prepare_upload(ase)
        # encrypt if necessary
        if ud.entity.is_encrypted and ud.entity.size > 0:
            # send iv through hmac if first chunk
            if offsets.chunk_num == 0:
                ud.hmac_data(ud.current_iv)
            # encrypt data
            if self._crypto_offload is None:
                # read data from file and encrypt
                data, _ = ud.read_data(offsets)
                encdata = blobxfer.operations.crypto.aes_cbc_encrypt_data(
                    ud.entity.encryption_metadata.symmetric_key,
                    ud.current_iv, data, offsets.pad)
                # send encrypted data through hmac
                ud.hmac_data(encdata)
                data = encdata
                # save last 16 encrypted bytes for next IV
                ud.current_iv = \
                    encdata[-blobxfer.models.crypto.AES256_BLOCKSIZE_BYTES:]
            else:  # noqa
                # crypto offload is not supported with AES256-CBC FullBlob
                raise NotImplementedError()
                # self._crypto_offload.add_encrypt_chunk(
                #     str(ud.local_path.absolute_path), offsets,
                #     ud.entity.encryption_metadata.symmetric_key,
                #     ud.current_iv)
                # encrypted data will be retrieved from a temp file once
                # retrieved from crypto queue
                # return_early = True
        else:
            data, newoffset = ud.read_data(offsets)
            # set new offset if stdin
            if newoffset is not None:
                offsets = newoffset
        # re-enqueue for other threads to upload if not append
        if ud.entity.mode != blobxfer.models.azure.StorageModes.Append:
            self._upload_queue.put(ud)
        # no data can be returned on stdin uploads
        if ud.local_path.use_stdin and not data:
            return
        # add data to transfer queue
        with self._transfer_lock:
            self._transfer_set.add(
                blobxfer.operations.upload.Uploader.create_unique_transfer_id(
                    ud.local_path, ud.entity, offsets))
        self._transfer_queue.put((ud, ud.entity, offsets, data))
        # iterate replicas
        if blobxfer.util.is_not_empty(ud.entity.replica_targets):
            for ase in ud.entity.replica_targets:
                with self._transfer_lock:
                    self._transfer_set.add(
                        blobxfer.operations.upload.Uploader.
                        create_unique_transfer_id(ud.local_path, ase, offsets)
                    )
                self._transfer_queue.put((ud, ase, offsets, data))
301552
def process(self):
    
    client = self._get_client_by_hostname(self.host)
    self._await_flow(client, self.flow_id)
    collected_flow_data = self._download_files(client, self.flow_id)
    if collected_flow_data:
      print('{0:s}: Downloaded: {1:s}'.format(
          self.flow_id, collected_flow_data))
      fqdn = client.data.os_info.fqdn.lower()
      self.state.output.append((fqdn, collected_flow_data))
177711
def _wait_for_disk_threads(self, terminate):
        # type: (Downloader, bool) -> None
        
        if terminate:
            self._download_terminate = terminate
        for thr in self._disk_threads:
            blobxfer.util.join_thread(thr)
372270
def __on_download_progress_update(self, blocknum, blocksize, totalsize):
        
        if not self.__show_download_progress:
            return

        readsofar = blocknum * blocksize
        if totalsize > 0:
            s = "\r%s / %s" % (size(readsofar), size(totalsize))
            sys.stdout.write(s)
            if readsofar >= totalsize:  # near the end
                sys.stderr.write("\r")
        else:  # total size is unknown
            sys.stdout.write("\rread %s" % (size(readsofar)))