MF - 1416 - Queue Abstraction for Mainflux & RabbitMQ Support

[rodneyosodo]

What does this do?

Changes the messaging broker from NATS to RabittMQ

Which issue(s) does this PR fix/relate to?

Resolves #1416

List any changes that modify/break current functionality

Have you included tests for your changes?

Not yet. Will include after approval

Did you document any new/modified functionality?

Not yet

Notes

N/A

[drasko]

@0x6f736f646f CI is failing. I restarted it, but there seems to be a problem with rabbitmq tests.

[rodneyosodo]

Working to resolve the issue

[manuio]

Looks like the indentation is worng here

[rodneyosodo]

Resolved

[manuio]

new line at the end of the file please

[rodneyosodo]

Resolved

[codecov-commenter]

Codecov Report

Merging #1562 (57dd8e3) into master (163c1b2) will increase coverage by 0.07%.
The diff coverage is 69.46%.

❗ Current head 57dd8e3 differs from pull request most recent head 74c1126. Consider uploading reports for the commit 74c1126 to get more accurate results

@@            Coverage Diff             @@
##           master    #1562      +/-   ##
==========================================
+ Coverage   68.93%   69.01%   +0.07%     
==========================================
  Files         136      138       +2     
  Lines       11058    11169     +111     
==========================================
+ Hits         7623     7708      +85     
- Misses       2788     2796       +8     
- Partials      647      665      +18     

Impacted Files	Coverage Δ
pkg/messaging/nats/publisher.go	52.38% <0.00%> (+26.06%)	⬆️
pkg/messaging/rabbitmq/pubsub.go	67.50% <67.50%> (ø)
pkg/messaging/rabbitmq/publisher.go	69.23% <69.23%> (ø)
pkg/messaging/nats/pubsub.go	68.49% <100.00%> (+3.43%)	⬆️
things/service.go	66.07% <0.00%> (+0.71%)	⬆️

📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

[dborovcanin]

...holds the implementation....

[rodneyosodo]

Done

[dborovcanin]

Is there any reason these constants are exported?

[rodneyosodo]

No reason. I am changing that

[rodneyosodo]

Done

[dborovcanin]

Please change the value of the routingKey.

[rodneyosodo]

Done

[dborovcanin]

There is no wg.Wait. What's the purpose of this WaitGroup?

[rodneyosodo]

Removed the waitgroup. The wait command was blocking

[dborovcanin]

Is this used?

[rodneyosodo]

Yes

[dborovcanin]

What for, what's the purpose of the done channel?

[rodneyosodo]

When we finish processing messages from the delivery channel we send a nil value

[dborovcanin]

Should this map be protected for concurrent read/write?

[rodneyosodo]

How?

[dborovcanin]

Usually by using a mutex.

[rodneyosodo]

Done

[dborovcanin]

There is no need to reimplement this in pubsub, you can use the implementation from the publisher.go:

type pubsub struct {
	publisher
	logger log.Logger
	queue  amqp.Queue
	subscriptions map[string]bool
	mutex sync.Mutex
}

[rodneyosodo]

Good idea

[rodneyosodo]

I am getting this error panic: runtime error: invalid memory address or nil pointer dereferenc. I have resorted to another approach

[dborovcanin]

Since this is rather a simple struct, consider using conn and ch instead.

[rodneyosodo]

Done

[rodneyosodo]

Done

[dborovcanin]

Is it OK to have an empty string for a consumer here?

[rodneyosodo]

An empty string will cause the library to generate a unique identity for each consumer. Since we have no way to id the subscriber, I think that is the best fit for now.

[dborovcanin]

How does canceling for an empty consumer work? Did you test subscribing with multiple consumers and canceling one of them (add that in test cases, as well)?

[rodneyosodo]

Initially, I thought it would generate the id for the exact consumer but after going through the code it doesn't as referenced here

[rodneyosodo]

If we do a ps.ch.close() we close the whole channel so we can't subscribe to it. The best solution to this is if we are able to generate UUID so as to id the consumers so we can cancel the respective consumers. The current implementation works fine.

[dborovcanin]

If you take publisher implementation from publisher.go that's already tested in publisher_test.go, you can have only Subscriber (with Subscribe and Unsubscribe) methods tested here.

[rodneyosodo]

Implemented

[dborovcanin]

Fix typos: sub topic -> subtopic and Unsusbcribe -> Unsubscribe.

[rodneyosodo]

Done

[drasko]

Needs variable name

[dborovcanin]

@drasko I suggested using the composition for reasons explained here.

[dborovcanin]

Instead of having a publisher field, consider using a composition of publisher like:

type pubsub struct {
	publisher
	logger        log.Logger
	mu            sync.Mutex
	queue         string
	subscriptions map[string]map[string]subscription
}

This will enable using ps.conn rather than ps.publisher.conn later on.

[rodneyosodo]

That is what we had initially but @drasko had suggested with name it.

I will stick with the earlier changes we had.

[dborovcanin]

Is this type correct? Since our message is binary, it looks like application/octet-stream is more suitable for us.

[rodneyosodo]

Yes application/octet-stream is more suitable

[dborovcanin]

Did you consider using topic kind? In that case, we have a queue per channel and the subtopic could be the routing key. This will have an impact on both publisher and subscriber implementation.

[rodneyosodo]

Yes. That will be a good approach but it poses come challenges because when we do a subscribe the topic preprocessing is done by the client compared to how it is implemented on the publish method.

[dborovcanin]

I understand that we'll have to disassemble the topic we created before calling Publish/Subscribe methods, but it's worth trying it because topic manipulation is rather simple. Of course, queue manipulation is also something to take care of (creating a queue if it doesn't exist and removing it if there are no messages and subscribers left). I'm only afraid this topic processing is going to be too cryptic - using the first part of the topic as a queue (name), and the rest as a subtopic (routingKey).

[rodneyosodo]

From this reference we need to have multiple queues. The current implementation I have pushed is that we have direct exchange rather than topic exchange as direct is faster than topic exchange. We will have a queue per subscription.
Another way to implement it is to have queues based on services such as a coap queue and an MQTT queue. But this makes us not have many queues.

[rodneyosodo]

Overview

RabbitMQ is somehow different to Nats and Kafka in that when a publisher is publishing a message on a topic they will publish the message to an exchange. The message will be published with a routing key. This routing key will direct the message to the different queues onto which the subscriber will consume the message. This is how the message flows

publisher -> exchange -> routingkey -> queue -> consumer -> subscriber


The publisher publishes the message to the broker.
The exchange receives the message being published and uses the routing key to route the message to a specific queue
The routing key is used to route the message to a specific queue
The queue temporarily stores messages for the consumer before consuming them

Exchange

When the exchange is declared we declare it with a specific kind. This means an exchange can either route messages based on topic, directly to queues or using message headers among other methods. We will be interested in the three routing techniques. The number of bindings increases the performance of topic routing. Fanout exchanges are very fast because there is no routing to process, but this changes if they are bound to a large number of queues. Direct exchanges are faster than topic exchanges if you don't need the wild card. We picked direct method as it strikes a balance between performance and reliability.

Possible implementations

One exchange one queue

• If you only have one queue, you'll end up repeating the logic of deciding which code should process the message, when RabbitMQ can do it for you with routing.
• RabbitMQ prefers many queues over a single large one
• This option is ruled out

One exchange multiple queues

• This is the most viable solution currently
• This is because we will have single consumers on one queue
• The advantage of a single topic exchange is that you get the benefits of both a Direct Exchange and a Fanout Exchange depending on the binding key used.
• Further, configuration changes are easy to achieve and can often be done without disrupting any system processing at all

Multiple exchanges one queue

• This option is ruled out based on having a single queue

Multiple exchanges multiple queues

• This is the most preferred way of doing it as the system is highly decoupled.
• If we cross the boundary and have many exchanges and many queues we are at the risk of the processes taking up more memory which may be significant but this really depends
• If the msg.channel is not provided the message won't be consumed as we can subscribe to an empty topic

Wayforward

• Currently, we have implemented Direct as the exchange type. So if a consumer needs to subscribe to multiple topics we would change this implementation to the exchange having topic as the exchange type
• We have not explored exchange-to-exchange architectures

For more reference check out this article

[manuio]

I'd create errors in a package for nats and rabbitmq and use them in the code and tests

[dborovcanin]

No need to cast, data is already []byte.

[dborovcanin]

Fix typo Susbcribe. Check for other typos as well.

[dborovcanin]

What's the purpose of this and is it even used?

[rodneyosodo]

It is not being used

[dborovcanin]

Is it safer to use topic instead of queue.Name? Is it possible that the broker will change the name passed in QueueDeclare?

[rodneyosodo]

The broker doesn't change the passed name

[dborovcanin]

Is this used?

[rodneyosodo]

exchangeKind no. The others yes.

[dborovcanin]

Should we really cancel unconditionally?

[rodneyosodo]

From the docs, this is what it says about the Cancel method

Only use this method to cleanly stop receiving deliveries from the server and cleanly shut down the consumer chan identified by this tag. Using this method and waiting for the remaining messages to flush from the consumer chan will ensure all messages received on the network will be delivered to the receiver of your consumer chan.

I believe this is what we want to achieve

[dborovcanin]

We are, but we should first handle the internal logic, then try to cancel and handle the potential errors. What happens if we have the same consumer subscribed to multiple topics? Does it cancel them all (because it's the same consumer ID)?
For example, a nice solution would be to put this in the cancel method of subscriber when you subscribe, so that you don't even call it explicitly when unsubscribing. Something like:

	s[id] = subscription{
		cancel: func() error {
			if err := ps.ch.Cancel(id, false); err != nil {
				return err
			}
			return handler.Cancel()
		},
	}

[rodneyosodo]

RabbitMQ expects that all client IDs are unique.

The consumer is identified by a unique string and scoped for all consumers on this channel.

So when we cancel the consumer we have a unique id that will only cancel one consumer. For our case, since we are passing in the unique ID, we need to generate a unique ID for every consumer or to create multiple channels per topic. Creating numerous channels is discouraged as by doing so it consumes system resources and makes it more difficult to configure firewalls

[dborovcanin]

RabbitMQ expects that all client IDs are unique.

The consumer is identified by a unique string and scoped for all consumers on this channel.

So when we cancel the consumer we have a unique id that will only cancel one consumer. For our case, since we are passing in the unique ID, we need to generate a unique ID for every consumer or to create multiple channels per topic. Creating numerous channels is discouraged as by doing so it consumes system resources and makes it more difficult to configure firewalls

OK. I let you choose between generating an ID and using a combination of client ID and topic.

[rodneyosodo]

RabbitMQ expects that all client IDs are unique.

The consumer is identified by a unique string and scoped for all consumers on this channel.

So when we cancel the consumer we have a unique id that will only cancel one consumer. For our case, since we are passing in the unique ID, we need to generate a unique ID for every consumer or to create multiple channels per topic. Creating numerous channels is discouraged as by doing so it consumes system resources and makes it more difficult to configure firewalls

OK. I let you choose between generating an ID and using a combination of client ID and topic.

Which one do you suggest we use? I had started with using a combination of client ID and topic

[dborovcanin]

Reuse clientID from L104.

[dborovcanin]

You can still keep TestPublish and TestSubscribe with separate cases (but in this single file, that's OK).

[rodneyosodo]

Here we are subscribing so that we can receive the published messages

[dborovcanin]

typo Unsubscribe **from**.... Same for all unsubscribe cases.

[drasko]

LGTM