藥劑流量：管道多個參數

Question

我想寫一個Flow，看起來像這樣：

def perform(id) do 
    id 
    |> Flow.from_stage(max_demand: 10) 
    |> Flow.map(&download(&1)) 
    |> Flow.map(&process(&1)) 
    |> Flow.each(&cleanup(&1, &2)) 
    |> Flow.each(&respond(&1, &2)) 
    |> Flow.run 
end 

其他功能模塊是這樣的：

def download(id) do 
    Service.get_images(id) 
    id 
end 

def process(id) do 
    %Result{out: sys_out, status: _} = Porcelain.exec("osascript", 
    ["#{File.cwd!}/lib/script/test", "#{id}", "#{Application.get_env(:app, :server_env)}"] 
) 
    {id, sys_out} 
end 

def cleanup(id, files) do 
    for file <- String.split(files, " ") do 
    System.cmd("exiftool", ["#{File.cwd!}/#{file}"]) 
    end 
    {id, files} 
end 

def respond(id, files) do 
    files = String.split(files, " ") 
    System.cmd("curl", [" -d 'dowhatever=#{files[0]}&else=#{files[1]}' -k #{Application.get_env(:app, :api)}what/#{id}/notify"]) 
    :ok 
end 

但我不斷收到此錯誤：

** (FunctionClauseError) no function clause matching in Flow.each/2 
... 
(app) lib/app/processor.ex:18: App.Processor.perform/1 

行18是&cleanup/2行。我在這裏做錯了什麼？這感覺就像我沒有返回正確的價值觀莫名其妙......

Answer 1

process/1返回一個元組{id, sys_out}，而cleanup/2聲明具有元數。

遍歷元組的列表意味着迭代器必須是元數1.你應該分解你的元組內cleanup/1：

- def cleanup(id, files) do 
+ def cleanup({id, files}) do 

，並調用它像：

Flow.each(&cleanup/1) 

同樣適用到後續的respond。