Event-Driven Architecture

<h2 id="event-driven-architecture" class="position-relative d-flex align-items-center group"> Event-Driven Architecture <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-driven-architecture" aria-haspopup="dialog" aria-label="Share link: Event-Driven Architecture"> Share link </button> </h2><div id="headingShareModal" class="heading-share-modal" role="dialog" aria-modal="true" aria-labelledby="headingShareTitle" hidden> <div class="hsm-dialog" role="document"> <div class="hsm-header"> <h2 id="headingShareTitle" class="h6 mb-0 fw-bold">Share this section</h2> <button type="button" class="hsm-close" aria-label="Close"> </button> </div> <div class="hsm-body"> <label for="headingShareInput" class="form-label small text-muted mb-1 text-uppercase fw-bold" style="font-size: 0.7rem; letter-spacing: 0.5px;">Permalink</label> <div class="input-group mb-4 hsm-url-group"> <input id="headingShareInput" type="text" class="form-control font-monospace" readonly aria-readonly="true" style="font-size: 0.85rem;" /> <button class="btn btn-primary hsm-copy" type="button" aria-label="Copy" title="Copy"> </button> </div> <div class="small fw-bold mb-2 text-muted text-uppercase" style="font-size: 0.7rem; letter-spacing: 0.5px;">Share via</div> <div class="hsm-share-grid"> <a id="share-twitter" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> Twitter </a> <a id="share-linkedin" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> LinkedIn </a> <a id="share-facebook" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> Facebook </a> </div> </div> </div> </div> <style> .heading-share-modal { position: fixed; inset: 0; display: flex; justify-content: center; align-items: center; background: rgba(0, 0, 0, 0.6); z-index: 1050; padding: 1rem; backdrop-filter: blur(4px); -webkit-backdrop-filter: blur(4px); } .heading-share-modal[hidden] { display: none !important; } .hsm-dialog { max-width: 420px; width: 100%; background: var(--bs-body-bg, #fff); color: var(--bs-body-color, #212529); border: 1px solid var(--bs-border-color, rgba(0,0,0,0.1)); border-radius: 1rem; box-shadow: 0 25px 50px -12px rgba(0, 0, 0, 0.25); overflow: hidden; animation: hsm-fade-in 0.2s ease-out; } @keyframes hsm-fade-in { from { opacity: 0; transform: scale(0.95); } to { opacity: 1; transform: scale(1); } } [data-bs-theme="dark"] .hsm-dialog { background: #1e293b; border-color: rgba(255,255,255,0.1); color: #f8f9fa; } .hsm-header { display: flex; justify-content: space-between; align-items: center; padding: 1rem 1.5rem; border-bottom: 1px solid var(--bs-border-color, rgba(0,0,0,0.1)); background: rgba(0,0,0,0.02); } [data-bs-theme="dark"] .hsm-header { background: rgba(255,255,255,0.02); border-color: rgba(255,255,255,0.1); } .hsm-close { background: transparent; border: none; color: inherit; opacity: 0.5; padding: 0.25rem 0.5rem; border-radius: 0.25rem; font-size: 1.2rem; line-height: 1; transition: opacity 0.2s; } .hsm-close:hover { opacity: 1; } .hsm-body { padding: 1.5rem; } .hsm-url-group { display: flex !important; align-items: stretch; } .hsm-url-group .form-control { flex: 1; min-width: 0; margin: 0; background: var(--bs-secondary-bg, #f8f9fa); border-color: var(--bs-border-color, #dee2e6); border-top-right-radius: 0; border-bottom-right-radius: 0; height: 42px; } .hsm-url-group .btn { flex: 0 0 auto; margin: 0; margin-left: -1px; border-top-left-radius: 0; border-bottom-left-radius: 0; height: 42px; display: flex; align-items: center; justify-content: center; padding: 0 1.25rem; z-index: 2; } [data-bs-theme="dark"] .hsm-url-group .form-control { background: #0f172a; border-color: #334155; color: #e2e8f0; } .hsm-share-grid { display: flex; flex-direction: column; gap: 0.5rem; } .hsm-share-grid .btn { display: flex; align-items: center; justify-content: center; font-size: 0.9rem; padding: 0.6rem; border-color: var(--bs-border-color); width: 100%; } [data-bs-theme="dark"] .hsm-share-grid .btn { color: #e2e8f0; border-color: #475569; } [data-bs-theme="dark"] .hsm-share-grid .btn:hover { background: #334155; border-color: #cbd5e1; } </style> <script> (function(){ const modal = document.getElementById('headingShareModal'); if(!modal) return; const input = modal.querySelector('#headingShareInput'); const copyBtn = modal.querySelector('.hsm-copy'); const twitter = modal.querySelector('#share-twitter'); const linkedin = modal.querySelector('#share-linkedin'); const facebook = modal.querySelector('#share-facebook'); const closeBtn = modal.querySelector('.hsm-close'); let lastFocus=null; let trapBound=false; function buildUrl(id){ return window.location.origin + window.location.pathname + '#' + id; } function isOpen(){ return !modal.hasAttribute('hidden'); } function hydrate(id){ const url=buildUrl(id); input.value=url; const enc=encodeURIComponent(url); const text=encodeURIComponent(document.title); if(twitter) twitter.href=`https://twitter.com/intent/tweet?url=${enc}&text=${text}`; if(linkedin) linkedin.href=`https://www.linkedin.com/sharing/share-offsite/?url=${enc}`; if(facebook) facebook.href=`https://www.facebook.com/sharer/sharer.php?u=${enc}`; } function openModal(id){ lastFocus=document.activeElement; hydrate(id); if(!isOpen()){ modal.removeAttribute('hidden'); } requestAnimationFrame(()=>{ input.focus(); }); trapFocus(); } function closeModal(){ if(!isOpen()) return; modal.setAttribute('hidden',''); if(lastFocus && typeof lastFocus.focus==='function') lastFocus.focus(); } function copyCurrent(){ try{ navigator.clipboard.writeText(input.value).then(()=>feedback(true),()=>fallback()); } catch(e){ fallback(); } } function fallback(){ input.select(); try{ document.execCommand('copy'); feedback(true);}catch(e){ feedback(false);} } function feedback(ok){ if(!copyBtn) return; const icon=copyBtn.querySelector('i'); if(!icon) return; const prev=copyBtn.getAttribute('data-prev')||icon.className; if(!copyBtn.getAttribute('data-prev')) copyBtn.setAttribute('data-prev',prev); icon.className= ok ? 'fa-duotone fa-clipboard-check':'fa-duotone fa-circle-exclamation'; setTimeout(()=>{ icon.className=prev; },1800); } function handleShareClick(e){ e.preventDefault(); const btn=e.currentTarget; const id=btn.getAttribute('data-share-target'); if(id) openModal(id); } function bindShareButtons(){ document.querySelectorAll('.h-share').forEach(btn=>{ if(!btn.dataset.hShareBound){ btn.addEventListener('click', handleShareClick); btn.dataset.hShareBound='1'; } }); } bindShareButtons(); if(document.readyState==='loading'){ document.addEventListener('DOMContentLoaded', bindShareButtons); } else { requestAnimationFrame(bindShareButtons); } document.addEventListener('click', function(e){ const shareBtn=e.target.closest && e.target.closest('.h-share'); if(shareBtn && !shareBtn.dataset.hShareBound){ handleShareClick.call(shareBtn, e); } }, true); document.addEventListener('click', e=>{ if(e.target===modal) closeModal(); if(e.target.closest && e.target.closest('.hsm-close')){ e.preventDefault(); closeModal(); } if(copyBtn && (e.target===copyBtn || (e.target.closest && e.target.closest('.hsm-copy')))) { e.preventDefault(); copyCurrent(); } }); document.addEventListener('keydown', e=>{ if(e.key==='Escape' && isOpen()) closeModal(); }); function trapFocus(){ if(trapBound) return; trapBound=true; modal.addEventListener('keydown', f=>{ if(f.key==='Tab' && isOpen()){ const focusable=[...modal.querySelectorAll('a[href],button,input,textarea,select,[tabindex]:not([tabindex="-1"])')].filter(el=>!el.hasAttribute('disabled')); if(!focusable.length) return; const first=focusable[0]; const last=focusable[focusable.length-1]; if(f.shiftKey && document.activeElement===first){ f.preventDefault(); last.focus(); } else if(!f.shiftKey && document.activeElement===last){ f.preventDefault(); first.focus(); } } }); } if(closeBtn) closeBtn.addEventListener('click', e=>{ e.preventDefault(); closeModal(); }); })(); </script>Event-Driven Architecture (EDA) is a software design pattern where the flow of the program is determined by events. Geode’s support for event-driven patterns enables building reactive, scalable, and loosely coupled systems that respond to changes as they occur. <h3 id="event-driven-fundamentals" class="position-relative d-flex align-items-center group"> Event-Driven Fundamentals <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-driven-fundamentals" aria-haspopup="dialog" aria-label="Share link: Event-Driven Fundamentals"> Share link </button> </h3> <h4 id="what-are-events" class="position-relative d-flex align-items-center group"> What are Events? <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="what-are-events" aria-haspopup="dialog" aria-label="Share link: What are Events?"> Share link </button> </h4>Events represent facts about things that happened: State Change - User updated profile Action Performed - Order placed Time Elapsed - Subscription expired Threshold Crossed - Inventory below minimum Events are: <ul> <li>Immutable - Cannot be changed after creation</li> <li>Timestamped - When did it happen</li> <li>Attributed - Who/what caused it</li> <li>Typed - What kind of event</li> </ul> <h4 id="event-driven-vs-traditional" class="position-relative d-flex align-items-center group"> Event-Driven vs Traditional <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-driven-vs-traditional" aria-haspopup="dialog" aria-label="Share link: Event-Driven vs Traditional"> Share link </button> </h4><table> <thead> <tr> <th>Aspect</th> <th>Traditional</th> <th>Event-Driven</th> </tr> </thead> <tbody> <tr> <td>Coupling</td> <td>Tight</td> <td>Loose</td> </tr> <tr> <td>Flow</td> <td>Synchronous</td> <td>Asynchronous</td> </tr> <tr> <td>Scalability</td> <td>Vertical</td> <td>Horizontal</td> </tr> <tr> <td>Responsiveness</td> <td>Request-driven</td> <td>Event-driven</td> </tr> <tr> <td>Complexity</td> <td>Linear</td> <td>Distributed</td> </tr> </tbody> </table> <h3 id="geode-event-system" class="position-relative d-flex align-items-center group"> Geode Event System <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="geode-event-system" aria-haspopup="dialog" aria-label="Share link: Geode Event System"> Share link </button> </h3> <h4 id="defining-event-handlers" class="position-relative d-flex align-items-center group"> Defining Event Handlers <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="defining-event-handlers" aria-haspopup="dialog" aria-label="Share link: Defining Event Handlers"> Share link </button> </h4>Register handlers for graph events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Handle user creation CREATE EVENT HANDLER new_user_handler ON INSERT Person EXECUTE GQL INSERT (n:Notification { type: 'welcome', user_id: NEW.id, created: NOW() }); -- Handle relationship creation CREATE EVENT HANDLER friendship_handler ON INSERT FRIEND EXECUTE GQL INSERT (e:Event { type: 'friendship_created', user1_id: NEW.from_id, user2_id: NEW.to_id, timestamp: NOW() }); </code></pre></div> <h4 id="event-processing" class="position-relative d-flex align-items-center group"> Event Processing <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-processing" aria-haspopup="dialog" aria-label="Share link: Event Processing"> Share link </button> </h4>Process events asynchronously: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from geode_client import Client class EventProcessor: def __init__(self, client): self.client = client self.handlers = {} def register_handler(self, event_type, handler): '''Register event handler''' if event_type not in self.handlers: self.handlers[event_type] = [] self.handlers[event_type].append(handler) async def process_events(self): '''Process events from stream''' async with self.client.subscribe_events() as stream: async for event in stream: await self.dispatch_event(event) async def dispatch_event(self, event): '''Dispatch event to handlers''' event_type = event['type'] if event_type in self.handlers: for handler in self.handlers[event_type]: try: await handler(event) except Exception as e: await self.handle_error(event, e) # Usage processor = EventProcessor(client) processor.register_handler('user.created', send_welcome_email) processor.register_handler('user.created', create_default_preferences) processor.register_handler('order.placed', process_payment) await processor.process_events() </code></pre></div> <h3 id="event-sourcing" class="position-relative d-flex align-items-center group"> Event Sourcing <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-sourcing" aria-haspopup="dialog" aria-label="Share link: Event Sourcing"> Share link </button> </h3> <h4 id="event-store" class="position-relative d-flex align-items-center group"> Event Store <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-store" aria-haspopup="dialog" aria-label="Share link: Event Store"> Share link </button> </h4>Store all events as source of truth: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Create event store CREATE EVENT_STORE account_events ( aggregate_id STRING, event_type STRING, event_data MAP<STRING, ANY>, version INTEGER, timestamp TIMESTAMP ) PARTITION BY aggregate_id; -- Append events INSERT INTO account_events VALUES ( 'account:123', 'AccountCreated', {'owner': 'alice', 'initial_balance': 1000}, 1, NOW() ); INSERT INTO account_events VALUES ( 'account:123', 'DepositMade', {'amount': 500, 'source': 'transfer'}, 2, NOW() ); </code></pre></div> <h4 id="rebuilding-state" class="position-relative d-flex align-items-center group"> Rebuilding State <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="rebuilding-state" aria-haspopup="dialog" aria-label="Share link: Rebuilding State"> Share link </button> </h4>Replay events to rebuild current state: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class AccountProjection: def __init__(self, account_id): self.account_id = account_id self.balance = 0 self.owner = None self.transactions = [] async def rebuild(self, client): '''Rebuild state from events''' events = await client.query( "SELECT * FROM account_events WHERE aggregate_id = $1 ORDER BY version", [self.account_id] ) for event in events: self.apply_event(event) def apply_event(self, event): '''Apply event to update state''' if event['event_type'] == 'AccountCreated': self.owner = event['event_data']['owner'] self.balance = event['event_data']['initial_balance'] elif event['event_type'] == 'DepositMade': self.balance += event['event_data']['amount'] self.transactions.append({ 'type': 'deposit', 'amount': event['event_data']['amount'], 'timestamp': event['timestamp'] }) elif event['event_type'] == 'WithdrawalMade': self.balance -= event['event_data']['amount'] self.transactions.append({ 'type': 'withdrawal', 'amount': event['event_data']['amount'], 'timestamp': event['timestamp'] }) </code></pre></div> <h3 id="cqrs-pattern" class="position-relative d-flex align-items-center group"> CQRS Pattern <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="cqrs-pattern" aria-haspopup="dialog" aria-label="Share link: CQRS Pattern"> Share link </button> </h3> <h4 id="command-model" class="position-relative d-flex align-items-center group"> Command Model <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="command-model" aria-haspopup="dialog" aria-label="Share link: Command Model"> Share link </button> </h4>Handle writes (commands): <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class AccountCommandHandler: def __init__(self, client): self.client = client async def handle_create_account(self, command): '''Handle account creation command''' event = { 'aggregate_id': command['account_id'], 'event_type': 'AccountCreated', 'event_data': { 'owner': command['owner'], 'initial_balance': command['initial_balance'] }, 'version': 1, 'timestamp': datetime.now() } await self.client.append_event('account_events', event) await self.publish_event(event) async def handle_deposit(self, command): '''Handle deposit command''' # Validate if command['amount'] <= 0: raise ValueError("Amount must be positive") # Get current version version = await self.get_current_version(command['account_id']) # Create event event = { 'aggregate_id': command['account_id'], 'event_type': 'DepositMade', 'event_data': { 'amount': command['amount'], 'source': command.get('source', 'unknown') }, 'version': version + 1, 'timestamp': datetime.now() } await self.client.append_event('account_events', event) await self.publish_event(event) </code></pre></div> <h4 id="query-model" class="position-relative d-flex align-items-center group"> Query Model <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="query-model" aria-haspopup="dialog" aria-label="Share link: Query Model"> Share link </button> </h4>Handle reads (queries): <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class AccountQueryHandler: def __init__(self, client): self.client = client self.cache = {} async def get_account_balance(self, account_id): '''Query account balance''' if account_id in self.cache: return self.cache[account_id]['balance'] # Rebuild from events projection = AccountProjection(account_id) await projection.rebuild(self.client) # Cache result self.cache[account_id] = { 'balance': projection.balance, 'owner': projection.owner } return projection.balance async def get_transaction_history(self, account_id, limit=100): '''Query transaction history''' events = await self.client.query( '''SELECT * FROM account_events WHERE aggregate_id = $1 AND event_type IN ('DepositMade', 'WithdrawalMade') ORDER BY timestamp DESC LIMIT $2''', [account_id, limit] ) return [ { 'type': event['event_type'], 'amount': event['event_data']['amount'], 'timestamp': event['timestamp'] } for event in events ] </code></pre></div> <h3 id="saga-pattern" class="position-relative d-flex align-items-center group"> Saga Pattern <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="saga-pattern" aria-haspopup="dialog" aria-label="Share link: Saga Pattern"> Share link </button> </h3>Coordinate distributed transactions: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class OrderSaga: def __init__(self, client): self.client = client async def execute(self, order_id): '''Execute order saga''' try: # Step 1: Reserve inventory await self.reserve_inventory(order_id) await self.publish_event('inventory_reserved', {'order_id': order_id}) # Step 2: Process payment await self.process_payment(order_id) await self.publish_event('payment_processed', {'order_id': order_id}) # Step 3: Ship order await self.ship_order(order_id) await self.publish_event('order_shipped', {'order_id': order_id}) except Exception as e: # Compensating transactions await self.compensate(order_id, e) async def compensate(self, order_id, error): '''Roll back completed steps''' await self.cancel_shipment(order_id) await self.refund_payment(order_id) await self.release_inventory(order_id) await self.publish_event('order_failed', { 'order_id': order_id, 'reason': str(error) }) </code></pre></div> <h3 id="best-practices" class="position-relative d-flex align-items-center group"> Best Practices <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="best-practices" aria-haspopup="dialog" aria-label="Share link: Best Practices"> Share link </button> </h3> <h4 id="event-design" class="position-relative d-flex align-items-center group"> Event Design <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-design" aria-haspopup="dialog" aria-label="Share link: Event Design"> Share link </button> </h4><ul> <li>Events represent facts that happened</li> <li>Use past tense for event names</li> <li>Include all relevant context</li> <li>Keep events immutable</li> <li>Version event schemas</li> </ul> <h4 id="processing-guarantees" class="position-relative d-flex align-items-center group"> Processing Guarantees <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="processing-guarantees" aria-haspopup="dialog" aria-label="Share link: Processing Guarantees"> Share link </button> </h4><ul> <li>Design idempotent handlers</li> <li>Handle duplicate events</li> <li>Implement proper error handling</li> <li>Use compensation for failures</li> <li>Monitor processing lag</li> </ul> <h4 id="performance" class="position-relative d-flex align-items-center group"> Performance <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="performance" aria-haspopup="dialog" aria-label="Share link: Performance"> Share link </button> </h4><ul> <li>Batch event processing</li> <li>Use async/await patterns</li> <li>Implement caching strategically</li> <li>Scale consumers horizontally</li> <li>Partition event streams</li> </ul> <h3 id="related-topics" class="position-relative d-flex align-items-center group"> Related Topics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="related-topics" aria-haspopup="dialog" aria-label="Share link: Related Topics"> Share link </button> </h3><ul> <li><a href="/tags/cdc" >CDC</a> - Change data capture</li> <li><a href="/tags/streaming" >Streaming</a> - Event streaming</li> <li><a href="/tags/pubsub" >Pub/Sub</a> - Publish-subscribe patterns</li> </ul> <h3 id="learn-more" class="position-relative d-flex align-items-center group"> Learn More <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="learn-more" aria-haspopup="dialog" aria-label="Share link: Learn More"> Share link </button> </h3><ul> <li><a href="https://martinfowler.com/articles/201701-event-driven.html" aria-label="Event-Driven Architecture (Martin Fowler) – opens in new window" target="_blank" rel="noopener noreferrer" >Event-Driven Architecture (Martin Fowler) ↗ </a> </li> <li><a href="https://martinfowler.com/eaaDev/EventSourcing.html" aria-label="Event Sourcing – opens in new window" target="_blank" rel="noopener noreferrer" >Event Sourcing ↗ </a> </li> <li><a href="https://martinfowler.com/bliki/CQRS.html" aria-label="CQRS Pattern – opens in new window" target="_blank" rel="noopener noreferrer" >CQRS Pattern ↗ </a> </li> </ul> <h3 id="production-event-patterns" class="position-relative d-flex align-items-center group"> Production Event Patterns <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="production-event-patterns" aria-haspopup="dialog" aria-label="Share link: Production Event Patterns"> Share link </button> </h3> <h4 id="change-data-capture-integration" class="position-relative d-flex align-items-center group"> Change Data Capture Integration <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="change-data-capture-integration" aria-haspopup="dialog" aria-label="Share link: Change Data Capture Integration"> Share link </button> </h4>Stream database changes to external systems: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from geode_client import open_database import asyncio class CDCProcessor: def __init__(self, geode_url, output_handler): self.client = open_database(geode_url) self.output = output_handler async def stream_changes(self, labels=None): """Poll ChangeLog entries and publish events.""" last_seen = "1970-01-01T00:00:00Z" async with self.client.connection() as conn: while True: result, _ = await conn.query( """ MATCH (e:ChangeLog) WHERE e.emitted_at > $since AND ($labels IS NULL OR e.label IN $labels) RETURN e.operation AS operation, e.label AS label, e.before AS before, e.after AS after, e.emitted_at AS emitted_at ORDER BY emitted_at """, {"since": last_seen, "labels": labels}, ) for row in result.rows: event = { "timestamp": row["emitted_at"].raw_value, "operation": row["operation"].raw_value, "label": row["label"].raw_value, "before": row["before"].raw_value, "after": row["after"].raw_value, } await self.output.publish(event) last_seen = row["emitted_at"].raw_value # Usage async def main(): processor = CDCProcessor( "quic://localhost:3141", KafkaOutput("kafka:9092", "geode-cdc") ) await processor.stream_changes(labels=['User', 'Order', 'Product']) </code></pre></div> <h4 id="event-sourcing-with-geode" class="position-relative d-flex align-items-center group"> Event Sourcing with Geode <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-sourcing-with-geode" aria-haspopup="dialog" aria-label="Share link: Event Sourcing with Geode"> Share link </button> </h4>Store events as source of truth: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class EventStore: def __init__(self, client): self.client = client async def append_event(self, aggregate_id, event_type, event_data, expected_version=None): """Append event with optimistic concurrency""" async with self.client.connection() as conn: await conn.begin() try: # Check version if specified if expected_version is not None: current, _ = await conn.query( """ MATCH (a:Aggregate {id: $id}) RETURN a.version as version """, {"id": aggregate_id}, ) row = current.rows[0] if current.rows else None if row and row["version"].as_int != expected_version: raise ConcurrencyError("Version mismatch") # Append event await conn.execute( """ MERGE (a:Aggregate {id: $aggregate_id}) ON CREATE SET a.version = 0 CREATE (e:Event { id: randomUUID(), aggregate_id: $aggregate_id, event_type: $event_type, event_data: $event_data, version: a.version + 1, timestamp: datetime() }) CREATE (a)-[:HAS_EVENT]->(e) SET a.version = a.version + 1 """, { "aggregate_id": aggregate_id, "event_type": event_type, "event_data": event_data, }, ) await conn.commit() except Exception: await conn.rollback() raise async def get_events(self, aggregate_id, from_version=0): """Get events for aggregate""" async with self.client.connection() as conn: result, _ = await conn.query( """ MATCH (a:Aggregate {id: $id})-[:HAS_EVENT]->(e:Event) WHERE e.version > $from_version RETURN e.event_type, e.event_data, e.version, e.timestamp ORDER BY e.version """, {"id": aggregate_id, "from_version": from_version}, ) return result.rows </code></pre></div> <h4 id="saga-orchestration" class="position-relative d-flex align-items-center group"> Saga Orchestration <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="saga-orchestration" aria-haspopup="dialog" aria-label="Share link: Saga Orchestration"> Share link </button> </h4>Coordinate distributed transactions: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class SagaOrchestrator: def __init__(self, client): self.client = client self.compensations = [] async def execute_saga(self, saga_id, steps): """Execute saga with compensation on failure""" async with self.client.connection() as conn: await conn.begin() # Create saga record await conn.execute( """ CREATE (s:Saga { id: $id, status: 'running', started_at: datetime() }) """, {"id": saga_id}, ) try: # Execute each step for i, step in enumerate(steps): await self.execute_step(conn, saga_id, i, step) # Mark success await conn.execute( """ MATCH (s:Saga {id: $id}) SET s.status = 'completed', s.completed_at = datetime() """, {"id": saga_id}, ) await conn.commit() except Exception as e: # Rollback with compensations await self.compensate(conn, saga_id) await conn.execute( """ MATCH (s:Saga {id: $id}) SET s.status = 'failed', s.error = $error """, {"id": saga_id, "error": str(e)}, ) await conn.commit() raise async def execute_step(self, conn, saga_id, step_num, step): """Execute single saga step""" # Record step await conn.execute(""" MATCH (s:Saga {id: $saga_id}) CREATE (step:SagaStep { saga_id: $saga_id, step_num: $step_num, action: $action, status: 'running' }) CREATE (s)-[:HAS_STEP]->(step) """, { "saga_id": saga_id, "step_num": step_num, "action": step['action'] }) # Execute step action execute_fn = step['execute'] await execute_fn(conn) # Store compensation self.compensations.append(step.get('compensate')) # Mark step complete await conn.execute(""" MATCH (step:SagaStep {saga_id: $saga_id, step_num: $step_num}) SET step.status = 'completed' """, {"saga_id": saga_id, "step_num": step_num}) async def compensate(self, conn, saga_id): """Execute compensating transactions""" for compensate in reversed(self.compensations): if compensate: await compensate(conn) </code></pre></div> <h4 id="real-time-event-processing" class="position-relative d-flex align-items-center group"> Real-Time Event Processing <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="real-time-event-processing" aria-haspopup="dialog" aria-label="Share link: Real-Time Event Processing"> Share link </button> </h4>Process events with low latency: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class RealTimeEventProcessor: def __init__(self, client): self.client = client self.handlers = {} def register_handler(self, event_type, handler): """Register event handler""" if event_type not in self.handlers: self.handlers[event_type] = [] self.handlers[event_type].append(handler) async def process_stream(self): """Process event stream in real-time""" async for event in self.client.subscribe_events(): # Dispatch to handlers await self.dispatch(event) async def dispatch(self, event): """Dispatch event to registered handlers""" event_type = event['type'] if event_type in self.handlers: # Execute handlers in parallel tasks = [ handler(event) for handler in self.handlers[event_type] ] await asyncio.gather(*tasks) # Usage processor = RealTimeEventProcessor(client) # Register handlers @processor.register_handler('UserCreated') async def send_welcome_email(event): user_email = event['data']['email'] await email_service.send(user_email, "Welcome!") @processor.register_handler('OrderPlaced') async def update_inventory(event): product_id = event['data']['product_id'] quantity = event['data']['quantity'] async with client.connection() as conn: await conn.execute( """ MATCH (p:Product {id: $id}) SET p.stock = p.stock - $qty """, {"id": product_id, "qty": quantity}, ) # Start processing await processor.process_stream() </code></pre></div>Browse the tagged content below to discover event-driven architecture patterns and real-time processing guides for Geode.

Popular

Related Articles