1"use strict";(self.webpackChunkcentrifugal_dev=self.webpackChunkcentrifugal_dev||[]).push([[3179],{28453:(e,n,s)=>{s.d(n,{R:()=>t,x:()=>c});var o=s(96540);const r={},i=o.createContext(r);function t(e){const n=o.useContext(i);return o.useMemo((function(){return"function"==typeof e?e(n):{...n,...e}}),[n,e])}function c(e){let n;return n=e.disableParentContext?"function"==typeof e.components?e.components(r):e.components||r:t(e.components),o.createElement(i.Provider,{value:n},e.children)}},78066:(e,n,s)=>{s.r(n),s.d(n,{assets:()=>a,contentTitle:()=>c,default:()=>h,frontMatter:()=>t,metadata:()=>o,toc:()=>l});const o=JSON.parse('{"id":"server/consumers","title":"Built-in API command async consumers","description":"In server API chapter we\'ve shown how to execute various Centrifugo server API commands (publish, broadcast, etc.) over HTTP or GRPC. In many cases you will call those APIs from your application business logic synchronously. But to deal with temporary network and availability issues, and achieve reliable execution of API commands upon changes in your primary application database you may want to use queuing techniques and call Centrifugo API asynchronously.","source":"@site/versioned_docs/version-5/server/consumers.md","sourceDirName":"server","slug":"/server/consumers","permalink":"/docs/5/server/consumers","draft":false,"unlisted":false,"editUrl":"https://github.com/centrifugal/centrifugal.dev/edit/main/versioned_docs/version-5/server/consumers.md","tags":[],"version":"5","frontMatter":{"id":"consumers","sidebar_label":"Async consumers","title":"Built-in API command async consumers"},"sidebar":"Guides","previous":{"title":"Engines and scalability","permalink":"/docs/5/server/engines"},"next":{"title":"History and recovery","permalink":"/docs/5/server/history_and_recovery"}}');var r=s(74848),i=s(28453);const t={id:"consumers",sidebar_label:"Async consumers",title:"Built-in API command async consumers"},c=void 0,a={},l=[{value:"Supported consumers",id:"supported-consumers",level:2},{value:"How it works",id:"how-it-works",level:2},{value:"Common consumer options",id:"common-consumer-options",level:2},{value:"PostgreSQL outbox consumer",id:"postgresql-outbox-consumer",level:2},{value:"PostgreSQL consumer options",id:"postgresql-consumer-options",level:3},{value:"Kafka consumer",id:"kafka-consumer",level:2},{value:"Kafka consumer options",id:"kafka-consumer-options",level:3},{value:"Kafka TLS options",id:"kafka-tls-options",level:3}];function d(e){const n={a:"a",admonition:"admonition",code:"code",h2:"h2",h3:"h3",li:"li",p:"p",pre:"pre",strong:"strong",ul:"ul",...(0,i.R)(),...e.components};return(0,r.jsxs)(r.Fragment,{children:[(0,r.jsxs)(n.p,{children:["In ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api",children:"server API"})," chapter we've shown how to execute various Centrifugo server API commands (publish, broadcast, etc.) over HTTP or GRPC. In many cases you will call those APIs from your application business logic synchronously. But to deal with temporary network and availability issues, and achieve reliable execution of API commands upon changes in your primary application database you may want to use queuing techniques and call Centrifugo API asynchronously."]}),"\n",(0,r.jsx)(n.p,{children:"Asynchronous delivery of real-time events upon changes in primary database may be done is several ways. Some companies use transactional outbox pattern, some using techniques like Kafka Connect with CDC (Change Data Capture) approach. The fact Centrifugo provides API allows users to implement any of those techniques and build worker which will send API commands to Centrifugo reliably."}),"\n",(0,r.jsx)(n.p,{children:"But Centrifugo also provides some built-in asynchronous consumers to simplify the integration process."}),"\n",(0,r.jsx)(n.h2,{id:"supported-consumers",children:"Supported consumers"}),"\n",(0,r.jsx)(n.p,{children:"The following built-in async consumers are available at this point:"}),"\n",(0,r.jsxs)(n.ul,{children:["\n",(0,r.jsxs)(n.li,{children:["Consumer ",(0,r.jsx)(n.a,{href:"#postgresql-outbox-consumer",children:"from PostgreSQL outbox table"})," (since Centrifugo v5.2.0)"]}),"\n",(0,r.jsxs)(n.li,{children:["Consumer ",(0,r.jsx)(n.a,{href:"#kafka-consumer",children:"from Kafka topics"})," (since Centrifugo v5.2.0)"]}),"\n"]}),"\n",(0,r.jsx)(n.h2,{id:"how-it-works",children:"How it works"}),"\n",(0,r.jsxs)(n.p,{children:["Consumers expect to consume messages which represent Centrifugo ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api",children:"server API commands"}),". I.e. while in synchronous server API you are using HTTP or GRPC to send commands \u2013 with asynchronous consumers you are inserting API command to PostgreSQL outbox table, or delivering to Kafka topic \u2013 and it will be soon consumed and processed asynchronously by Centrifugo."]}),"\n",(0,r.jsxs)(n.p,{children:["Async consumers only process commands which modify state \u2013 such as ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api#publish",children:"publish"}),", ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api#broadcast",children:"broadcast"}),", ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api#unsubscribe",children:"unsubscribe"}),", ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api#disconnect",children:"disconnect"}),", etc. Sending read commands for async execution simply does not make any sense and they will be ignored. Also, ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api#batch",children:"batch"})," method is not supported."]}),"\n",(0,r.jsxs)(n.p,{children:["Centrifugo ",(0,r.jsx)(n.strong,{children:"only supports JSON payloads for asynchronous commands coming to consumers for now"}),". If you need binary format \u2013 reach out with your use case."]}),"\n",(0,r.jsxs)(n.p,{children:["If Centrifugo encounters an error while processing consumed messages \u2013 then internal errors will be retried, all other errors logged on ",(0,r.jsx)(n.code,{children:"error"})," level \u2013 and the message will be marked as processed. The processing logic for ",(0,r.jsx)(n.a,{href:"/docs/5/server/server_api#broadcast",children:"broadcast"})," API is special: if any of the publications to any channel from broadcast ",(0,r.jsx)(n.code,{children:"channels"})," array failed \u2013 then the entire broadcast command will be retried. To prevent duplicate messages being published during such retries \u2013 consider using ",(0,r.jsx)(n.code,{children:"idempotency_key"})," in the broadcast command."]}),"\n",(0,r.jsx)(n.admonition,{type:"tip",children:(0,r.jsxs)(n.p,{children:["Our ",(0,r.jsx)(n.a,{href:"/docs/5/tutorial/outbox_cdc",children:"Chat/Messenger tutorial"})," shows PostgreSQL outbox and Kafka consumer in action. It also shows techniques to avoid duplicate messages (idempotent publications) and deal with late message delivery (idempotent processing on client side). Whether you need those techniques \u2013 depends on the nature of app. Various real-time features may require different ways of sending real-time events. Both synchronous API calls and async calls have its own advantages and trade-offs. We also talk about this in ",(0,r.jsx)(n.a,{href:"/blog/2023/08/19/asynchronous-message-streaming-to-centrifugo-with-benthos",children:"Asynchronous message streaming to Centrifugo with Benthos"})," blog post."]})}),"\n",(0,r.jsx)(n.h2,{id:"common-consumer-options",children:"Common consumer options"}),"\n",(0,r.jsxs)(n.p,{children:["Consumers can be set in the configuration using ",(0,r.jsx)(n.code,{children:"consumers"})," array:"]}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-json",children:'{\n ...\n "consumers": [\n {\n "name": "xxx",\n "type": "postgresql",\n "postgresql": {...}\n },\n {\n "name": "yyy",\n "type": "kafka",\n "kafka": {...}\n },\n ]\n}\n'})}),"\n",(0,r.jsx)(n.p,{children:"On top level each consumer object has the following fields:"}),"\n",(0,r.jsxs)(n.ul,{children:["\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"name"})," - string (required), described name of consumer. Must be unique for each consumer and match the regex ",(0,r.jsx)(n.code,{children:"^[a-zA-Z0-9_]{2,}"})," - i.e. latin symbols, digits and underscores and be at least 2 symbols. This name will be used for logging purposes, metrics, also to override some options with environment variables."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"type"})," - string (required), type of consumer. At this point can be ",(0,r.jsx)(n.code,{children:"postgresql"})," or ",(0,r.jsx)(n.code,{children:"kafka"})]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"disabled"})," - boolean (default: ",(0,r.jsx)(n.code,{children:"false"}),"), when set to ",(0,r.jsx)(n.code,{children:"true"})," allows disabling the consumer"]}),"\n"]}),"\n",(0,r.jsxs)(n.p,{children:["To provide ",(0,r.jsx)(n.code,{children:"consumers"})," over environment variable provide ",(0,r.jsx)(n.code,{children:"CENTRIFUGO_CONSUMERS"})," var with JSON array serialized to string. It's also possible to override some specific consumer options over environment variables \u2013 see below."]}),"\n",(0,r.jsx)(n.h2,{id:"postgresql-outbox-consumer",children:"PostgreSQL outbox consumer"}),"\n",(0,r.jsxs)(n.p,{children:["Centrifugo can natively integrate with PostgreSQL table for ",(0,r.jsx)(n.a,{href:"https://microservices.io/patterns/data/transactional-outbox.html",children:"Transactional outbox"})," pattern. The table in PostgreSQL must have predefined format Centrifugo expects:"]}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-sql",children:"CREATE TABLE IF NOT EXISTS centrifugo_outbox (\n\tid BIGSERIAL PRIMARY KEY,\n\tmethod text NOT NULL,\n\tpayload JSONB NOT NULL,\n\tpartition INTEGER NOT NULL default 0,\n\tcreated_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL\n);\n"})}),"\n",(0,r.jsxs)(n.p,{children:["Then configure consumer of ",(0,r.jsx)(n.code,{children:"postgresql"})," type in Centrifugo config:"]}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-json",children:'{\n ...\n "consumers": [\n {\n "name": "my_postgresql_consumer",\n "type": "postgresql",\n "postgresql": {\n "dsn": "postgresql://user:password@localhost:5432/db",\n "outbox_table_name": "centrifugo_outbox",\n "num_partitions": 1,\n "partition_select_limit": 100,\n "partition_poll_interval": "300ms"\n }\n }\n ]\n}\n'})}),"\n",(0,r.jsx)(n.p,{children:"Here is how you can insert row in outbox table to publish into Centrifugo channel:"}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-SQL",children:'INSERT INTO centrifugo_outbox (method, payload, partition)\nVALUES (\'publish\', \'{"channel": "updates", "data": {"text": "Hello, world!"}}\', 0);\n'})}
1),"\n",(0,r.jsx)(n.p,{children:"Centrifugo supports LISTEN/NOTIFY mechanism of PostgreSQL to be notified about new data in the outbox table. To enable it you need first create a trigger in PostgreSQL:"}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-sql",children:"CREATE OR REPLACE FUNCTION centrifugo_notify_partition_change()\nRETURNS TRIGGER AS $$\nBEGIN\n PERFORM pg_notify('centrifugo_partition_change', NEW.partition::text);\n RETURN NEW;\nEND;\n$$ LANGUAGE plpgsql;\n\nCREATE OR REPLACE TRIGGER centrifugo_notify_partition_trigger\nAFTER INSERT ON chat_outbox\nFOR EACH ROW\nEXECUTE FUNCTION centrifugo_notify_partition_change();\n"})}),"\n",(0,r.jsxs)(n.p,{children:["And then update consumer config \u2013 add ",(0,r.jsx)(n.code,{children:'"partition_notification_channel"'})," option to it:"]}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-json",children:'{\n ...\n "consumers": [\n {\n "name": "my_postgresql_consumer",\n "type": "postgresql",\n "postgresql": {\n ...\n "partition_notification_channel": "centrifugo_partition_change"\n }\n }\n ]\n}\n'})}),"\n",(0,r.jsx)(n.h3,{id:"postgresql-consumer-options",children:"PostgreSQL consumer options"}),"\n",(0,r.jsxs)(n.ul,{children:["\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"dsn"})," - string (required), DSN to PostgreSQL database, ex. ",(0,r.jsx)(n.code,{children:'"postgresql://user:password@localhost:5432/db"'}),". To override ",(0,r.jsx)(n.code,{children:"dsn"})," over environment variables use ",(0,r.jsx)(n.code,{children:"CENTRIFUGO_CONSUMERS_POSTGRESQL_<CONSUMER_NAME>_DSN"}),"."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"outbox_table_name"})," - string (required), the name of outbox table in selected database, ex. ",(0,r.jsx)(n.code,{children:'"centrifugo_outbox"'})]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"num_partitions"})," - integer (default: ",(0,r.jsx)(n.code,{children:"1"}),"), the number of partitions to use. Centrifugo keeps strict order of commands per-partition by default. This option provides a way to create concurrent consumers each consuming from different partition of outbox table. Note, that partition numbers in start with ",(0,r.jsx)(n.code,{children:"0"}),", so when using ",(0,r.jsx)(n.code,{children:"1"})," as ",(0,r.jsx)(n.code,{children:"num_partitions"})," insert data with ",(0,r.jsx)(n.code,{children:"partition"})," == ",(0,r.jsx)(n.code,{children:"0"})," to the outbox table."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"partition_select_limit"})," - integer (default: ",(0,r.jsx)(n.code,{children:"100"}),") \u2013 max number of commands to select in one query to outbox table."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"partition_poll_interval"})," - duration (default: ",(0,r.jsx)(n.code,{children:'"300ms"'}),") - polling interval for each partition"]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"partition_notification_channel"})," - string (default: ",(0,r.jsx)(n.code,{children:'""'}),") - optional name of LISTEN/NOTIFY channel to trigger consuming upon data added to outbox partition."]}),"\n"]}),"\n",(0,r.jsx)(n.h2,{id:"kafka-consumer",children:"Kafka consumer"}),"\n",(0,r.jsx)(n.p,{children:"Another built-in consumer \u2013 is Kafka topics consumer. To configure Centrifugo to consume Kafka topic:"}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-json",children:' ...\n "consumers": [\n {\n "name": "my_kafka_consumer",\n "type": "kafka",\n "kafka": {\n "brokers": ["localhost:9092"],\n "topics": ["postgres.public.chat_cdc"],\n "consumer_group": "centrifugo"\n }\n }\n ]\n}\n'})}),"\n",(0,r.jsx)(n.p,{children:"Then simply put message in the following format to Kafka topic:"}),"\n",(0,r.jsx)(n.pre,{children:(0,r.jsx)(n.code,{className:"language-json",children:'{\n "method": "publish",\n "payload": {\n "channel": "mychannel",\n "data": {}\n }\n}\n'})}),"\n",(0,r.jsx)(n.p,{children:"\u2013 and it will be consumed by Centrifugo and reliably processed."}),"\n",(0,r.jsx)(n.h3,{id:"kafka-consumer-options",children:"Kafka consumer options"}),"\n",(0,r.jsxs)(n.ul,{children:["\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"brokers"}
1)," - array of strings (required), points Centrifugo to Kafka brokers. To override ",(0,r.jsx)(n.code,{children:"brokers"})," over environment variables use ",(0,r.jsx)(n.code,{children:"CENTRIFUGO_CONSUMERS_KAFKA_<CONSUMER_NAME>_BROKERS"})," \u2013 string with broker addresses separated by space."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"topics"})," - array of string (required), tells which topics to consume"]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"consumer_group"})," - string (required), sets the name of consumer group to use"]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"max_poll_records"})," - integer (default: ",(0,r.jsx)(n.code,{children:"100"}),") - sets the maximum number of records to fetch from Kafka during a single poll operation."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"sasl_mechanism"})," - SASL mechanism to use: ",(0,r.jsx)(n.code,{children:'"plain"'}),", ",(0,r.jsx)(n.code,{children:'"scram-sha-256"'})," (since 5.4.7), ",(0,r.jsx)(n.code,{children:'"scram-sha-512"'})," (since 5.4.7), ",(0,r.jsx)(n.code,{children:'"aws-msk-iam"'})," (since 5.4.7) are supported. Note, in case of ",(0,r.jsx)(n.code,{children:'"aws-msk-iam"'})," Centrifugo uses ",(0,r.jsx)(n.code,{children:"sasl_user"})," and ",(0,r.jsx)(n.code,{children:"sasl_password"})," options as ",(0,r.jsx)(n.code,{children:"access key"})," and ",(0,r.jsx)(n.code,{children:"secret key"})," when configuring AWS auth."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"sasl_user"})," - string, user for plain SASL auth. To override ",(0,r.jsx)(n.code,{children:"sasl_user"})," over environment variables use ",(0,r.jsx)(n.code,{children:"CENTRIFUGO_CONSUMERS_KAFKA_<CONSUMER_NAME>_SASL_USER"}),"."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"sasl_password"})," - string, password for plain SASL auth. To override ",(0,r.jsx)(n.code,{children:"sasl_password"})," over environment variables use ",(0,r.jsx)(n.code,{children:"CENTRIFUGO_CONSUMERS_KAFKA_<CONSUMER_NAME>_SASL_PASSWORD"}),"."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls"})," - boolean (default: ",(0,r.jsx)(n.code,{children:"false"}),") - enables TLS, if ",(0,r.jsx)(n.code,{children:"true"})," \u2013 then all the ",(0,r.jsx)(n.a,{href:"#kafka-tls-options",children:"TLS options"})," may be additionally set."]}),"\n"]}),"\n",(0,r.jsx)(n.h3,{id:"kafka-tls-options",children:"Kafka TLS options"}),"\n",(0,r.jsxs)(n.ul,{children:["\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_cert"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," representing the path on disk to the TLS certificate. This is typically used for the server certificate in TLS connections."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_key"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," representing the path on disk to the TLS key associated with the certificate."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_cert_pem"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," containing the PEM (Privacy Enhanced Mail) encoded TLS certificate."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_key_pem"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," containing the PEM encoded TLS key."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_root_ca"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," representing the path on disk to the root CA certificate. This is used to verify the server certificate."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_root_ca_pem"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," containing the PEM encoded root CA certificate."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_client_ca"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," representing the path on disk to the client CA certificate. This is used in mutual TLS"]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_client_ca_pem"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," containing the PEM encoded client CA certificate."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"tls_insecure_skip_verify"})," \u2013 ",(0,r.jsx)(n.code,{children:"boolean"})," indicating whether to skip verifying the server's certificate chain and host name. If ",(0,r.jsx)(n.code,{children:"true"}),", TLS accepts any certificate presented by the server and any host name in that certificate."]}),"\n",(0,r.jsxs)(n.li,{children:[(0,r.jsx)(n.code,{children:"server_name"})," \u2013 ",(0,r.jsx)(n.code,{children:"string"})," specifying the server name to use for server certificate verification if ",(0,r.jsx)(n.code,{children:"tls_insecure_skip_verify"})," is false. This is also used in the client's SNI (Server Name Indication) extension in TLS handshake."]}),"\n"]})]})}function h(e={}){const{wrapper:n}={...(0,i.R)(),...e.components};return n?(0,r.jsx)(n,{...e,children:(0,r.jsx)(d,{...e})}):d(e)}}}]);
Line numbers count LF bytes from the start of the resource, as the search results do. Vendor segments are library code the classifier recognised; they are stored but not indexed. Bytes are shown as Latin1 characters, one per byte.