Ballerina for Event-Driven Architecture (EDA)

public type Order readonly & record {
    int orderId;
    string productName;
    decimal price;
    boolean isValid;
};

listener kafka:Listener orderListener = new (kafka:DEFAULT_URL, {
    groupId: "order-group-id",
    topics: "order-topic"
});

service on orderListener {

    remote function onConsumerRecord(Order[] orders) returns error? {
        // The set of orders received by the service are processed one by one.
        check from Order 'order in orders
            where 'order.isValid
            do {
                log:printInfo(string `Received valid order for ${'order.productName}`);
            };
    }
}

// This service is for drivers to register and send locations.
service /taxi on taxiMgtListener {
    resource isolated function get [string name]() 
                        returns websocket:Service|websocket:UpgradeError {
        return new DriverService(name);
    }
}

isolated service class DriverService {
    *websocket:Service;

    final string driverName;

    public isolated function init(string username) {
        self.driverName = username;
    }

    remote isolated function onOpen(websocket:Caller caller) returns websocket:Error? {
        string welcomeMsg = 
                "Hi " + self.driverName + "! Your location will be shared with the riders";
        check caller->writeMessage(welcomeMsg);
        
        broadcast("Driver " + self.driverName + " ready for a ride");
    }

    // 'onMessage' remote function will receive the location updates from drivers.
    remote function onMessage(websocket:Caller caller, string location) returns websocket:Error? {
        worker broadcast returns error? {
            // Broadcast the live locations to registered riders.
            string locationUpdateMsg = self.driverName + " updated the location " + location;
            broadcast(locationUpdateMsg);
        }
    }

    remote isolated function onClose(websocket:Caller caller, int statusCode, string reason) {
        io:println(self.driverName + " got disconnected");
    }
}

configurable string username = ?;
configurable string password = ?;

listener email:ImapListener emailListener = new ({
    host: "imap.email.com",
    username,
    password
});

service "observer" on emailListener {

    // When an email is successfully received, the `onMessage` method is called.
    remote function onMessage(email:Message email) {
        log:printInfo("Received an email", subject = email.subject, content = email?.body);
    }

    // When an error occurs during the email poll operations, the `onError` method is called.
    remote function onError(email:Error emailError) {
        log:printError(emailError.message(), stackTrace = emailError.stackTrace());
    }

    // When the listener is closed, the `onClose` method is called.
    remote function onClose(email:Error? closeError) {
        if closeError is email:Error {
            log:printInfo(closeError.message(), stackTrace = closeError.stackTrace());
        }
    }
}