Mqtt Subscriber / Initialization

MrPec · October 20, 2022, 6:23am

Hi Guys,

Imagine I write a custom app and I want to give the user the possibility to subscribe to various events on various brokers. My idea is to create a custom doctype in which the user can specify all credentials, the topic etc for a subscription the user wants to have. On save of that document I spawn a new mqtt client (or stop an existing), trying to subscribe to the given topic in a loop until the document is deleted or changed.

Is that the way to go? Or where would you put the code to startup the mqtt client?
I was unable to find the existing event doctype of Frappe, maybe its done in a similar way there?

Thanks!

Alex

revant_one · October 20, 2022, 7:01pm

each client will be separate process running.

it can be done through custom bench command.

it cannot be done through gunicorn/python process itself.

i can think of creating a worker pod/container for instance of doctype. creds can be passed as env vars. do it through container orchestration api and keep track of pod==mqtt-client-doc

MrPec · October 21, 2022, 6:41am

Thanks for your reply.
Re: Custom bench command, you mean like your mqtt library does?
(GitHub - revant/connect_mqtt)

On top of that - ENV to pass parameters, then Popen bench … mqtt_subscribe, and adopt your lib. Add PID to the document that spawned the subprocess in order to kill&restart, right?

When app initializes, iterate over doctypes that are enabled and startup with Popen N clients.

Is that correct? Thanks

revant_one · October 21, 2022, 11:19am

yes

That is simple for proof of concept. Just Popen bench won’t be sufficient in production. I think you’ll need to involve supervisor or some kind of process manager if you’re not using containers.

You won’t need this if you use containers api. Instead of Popen bench start the container with custom command,

github.com

frappe/frappe_docker/blob/82cc15253a5ee54b98fe4ea14f256b361f82a217/pwd.yml#L110-L119


      
          queue-long:
            image: frappe/erpnext-worker:v14.3.1
            deploy:
              restart_policy:
                condition: on-failure
            command:
              - bench
              - worker
              - --queue
              - long

or,

github.com

frappe/helm/blob/2ef742cca761d68e3b43d5691f6ef810bcf1d7d2/erpnext/templates/deployment-worker-long.yaml#L48-L58


      
          containers:
            - name: long
              {{- if .Values.worker.long.envVars }}
              env:
                {{- toYaml .Values.worker.long.envVars | nindent 12 }}
              {{- end }}
              args:
              - bench
              - worker
              - --queue
              - long

If you choose containers / pods then restarts and process management is handled for you by container engine or orchestrator, you just need to use the api to communicate with it.

MrPec · October 22, 2022, 5:33am

Thanks Revant, I’ll try your suggestion and see how far I come. As I have limited experience with docker I’ll start with a POC and move on from there.