ActionBlock에 Post한게 큐에서 대기하다 처리될때까지 await으로 기다린 후 결과를 받는 패턴을 객체화 시킴.

https://gall.dcinside.com/board/view?id=programming&no=2079638

이패턴 맘에 안들어서 다른 남이만든 기능을 찾다가 포기하고 정리함.


public class FuncBlock<I,O> : IDataflowBlock

{

readonly ActionBlock<bag> a;

IDataflowBlock ia => a;

public Task Completion => ia.Completion;

public void Complete() => ia.Complete();

public void Fault(Exception ex) => ia.Fault(ex);


class bag

{

public I input;

public readonly TaskCompletionSource<O> when = new TaskCompletionSource<O>();

public void set(Func<I, O> f) { try { when.SetResult(f(input)); } catch (Exception ex) { when.TrySetException(ex); } }

public async Task set(Func<I, Task<O>> f) { try { when.SetResult(await f(input)); } catch (Exception ex) { when.TrySetException(ex); } }

}


public FuncBlock(Func<I, O> f) => a = new ActionBlock<bag>(b => b.set(f));

public FuncBlock(Func<I, Task<O>> f) => a = new ActionBlock<bag>(b => b.set(f));


public async Task<O> Invoke(I input)

{

var b = new bag { input = input };

if (a.Post(b))

{

var w = b.when.Task;

await Task.WhenAny(w, Completion);

if (w.IsCompleted)

return await w;

}

throw new Exception($"block is {Completion.Status}");

}

}


테스트 안해봐서 버그있어도 몰름



+ 단순 Thread단위의 호출순서를 보장하려면 Task.Run 대신

TaskFactory<int> tf = new TaskFactory<int>(new ConcurrentExclusiveSchedulerPair().ExclusiveScheduler);

tf.StartNew(() => { /*do something*/ });

일케 하면 tf끼리는 서로 동기적으로 실행되게됨. 근데 그건 Thread단위의 동기를 의미하고 Task단위로는 await걸리는 순간 다 스까짐.