Packages
ex_aws
1.0.0-rc.1
2.7.0
2.6.1
2.6.0
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.10
2.2.9
2.2.8
2.2.7
2.2.6
2.2.5
2.2.4
2.2.3
2.2.2
2.2.1
2.2.0
2.1.9
2.1.8
2.1.7
2.1.6
2.1.5
2.1.4
2.1.3
2.1.2
2.1.1
2.1.0
2.0.2
2.0.1
2.0.0
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.0
1.0.0-rc.4
1.0.0-rc.3
1.0.0-rc.1
1.0.0-beta3
1.0.0-beta2
1.0.0-beta1
1.0.0-beta0
0.5.0
0.4.19
0.4.18
0.4.17
0.4.15
0.4.14
0.4.13
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.1
0.3.0
0.2.0
0.1.2
0.1.1
0.1.0
0.0.5
0.0.4
0.0.3
AWS client for Elixir. Currently supports Dynamo, DynamoStreams, EC2, Firehose, Kinesis, KMS, Lambda, RRDS, Route53, S3, SES, SNS, SQS, STS and others.
Current section
Files
Jump to
Current section
Files
lib/ex_aws/s3/upload.ex
defmodule ExAws.S3.Upload do
@moduledoc """
Represents an AWS S3 Multipart Upload operation
## Examples
```
"path/to/big/file"
|> S3.Upload.stream_file!
|> S3.upload("my-bucket", "path/on/s3")
|> ExAws.request! #=> {:ok, :done}
```
"""
@enforce_keys ~w(bucket path src)a
defstruct [
:src,
:bucket,
:path,
:upload_id,
opts: [],
service: :s3,
]
@type t :: %__MODULE__{
src: Enumerable.t,
bucket: binary,
path: binary,
upload_id: binary,
opts: Keyword.t,
service: :s3
}
def complete!(parts, op, config) do
ExAws.S3.complete_multipart_upload(op.bucket, op.path, op.upload_id, Enum.sort_by(parts, &elem(&1, 0)))
|> ExAws.request!(config)
end
def initialize!(op, config) do
%{body: %{upload_id: upload_id}} =
ExAws.S3.initiate_multipart_upload(op.bucket, op.path, op.opts)
|> ExAws.request!(config)
%{op | upload_id: upload_id}
end
@doc """
Open a file stream for use in an upload.
Chunk size must be at least 5 MiB. Defaults to 5 MiB
"""
@spec stream_file(path :: binary) :: File.Stream.t
@spec stream_file(path :: binary, opts :: [chunk_size: pos_integer]) :: File.Stream.t
def stream_file(path, opts \\ []) do
path
|> File.stream!([:raw, :read_ahead, :binary], opts[:chunk_size] || 5 * 1024 * 1024)
end
@doc """
Upload a chunk for an operation.
The first argument is a tuple with the binary contents of the chunk, and a
positive integer index indicating which chunk it is. It will return this index
along with the `etag` response from AWS necessary to complete the multipart upload.
"""
@spec upload_chunk!({binary, pos_integer}, t, ExAws.Config.t) :: {pos_integer, binary}
def upload_chunk!({chunk, i}, op, config) do
%{headers: headers} = ExAws.S3.upload_part(op.bucket, op.path, op.upload_id, i, chunk, op.opts)
|> ExAws.request!(config)
{_, etag} = List.keyfind(headers, "ETag", 0)
{i, etag}
end
end
defimpl ExAws.Operation, for: ExAws.S3.Upload do
alias Experimental.Flow
alias ExAws.S3.Upload
def perform(op, config) do
op = Upload.initialize!(op, config)
Flow.new(stages: op.opts[:max_concurrency] || 4, max_demand: 2)
|> Flow.from_enumerable(op.src)
|> Flow.map(&Upload.upload_chunk!(&1, op, config))
|> Enum.to_list
|> Upload.complete!(op, config)
{:ok, :done}
end
def stream!(_, _), do: raise "not implemented"
end