Rewrite InMemoryBroker over aiochannel

[hugiron]

trafficstars

[codecov-commenter]

Codecov Report

Merging #92 (b0e6b49) into master (318b25b) will increase coverage by 4.61%. The diff coverage is 100.00%.

:mega: This organization is not using Codecov’s GitHub App Integration. We recommend you install it so Codecov can continue to function properly for your repositories. Learn more

@@            Coverage Diff             @@
##           master      #92      +/-   ##
==========================================
+ Coverage   61.94%   66.55%   +4.61%     
==========================================
  Files          37       37              
  Lines         938      930       -8     
==========================================
+ Hits          581      619      +38     
+ Misses        357      311      -46     

Impacted Files	Coverage Δ
taskiq/brokers/inmemory_broker.py	86.66% <100.00%> (+24.40%)	:arrow_up:

... and 5 files with indirect coverage changes

:mega: We’re building smart automated test selection to slash your CI/CD build times. Learn more

[s3rius]

I don't think that this PR is going to be merged in the way it exists now. For two reasons.

First of all, where does all processing happens? I guess now you need to start async_listen_messages manually somewhere in your code. But before that InMemoryBroker supposed to be lightweight broker that can be started even without actual listening.

My proposal would be to create a separate broker that allows you to run code using memory channels. Maybe InMemoryChanneledBroker or you can call it as you want.

Also, you need to update docs. Because currently in docs there's no single word about you need to start async listen somewhere in your codebase explicitly.

[hugiron]

1. What is the difference between a memory broker in the current approach from just calling an function in thread/process pool?
2. Consistency breaks, as different brokers may have different listen logic

[s3rius]

1. There is a difference. Main idea is to replace broker inplace.

Before this PR you could write something like this:

broker = AioPikaBroker("...")

if os.environ.get("APP_ENV", "dev") == "dev":
	broker = InMemoryBroker()

Now you don't need to modify your code in any other places, you don't have to explicitly starting listening task somewhere. It's just a convenient interface to use if you don't want to start your distributed queues like rabbit or redis or any other queue. Also it's suitable for testing your functions that use tasks.

2. Yes, maybe. It doesn't listen to new tasks. The main point here is not to be consistent, but to be easy to use, with no additional actions required from users. We want to make library easy-to-use, but using this approach is harder that just replacing broker. Isn't it?