Module 7, Lesson 1.

Real-Time and Reactive Programming

 Introduction to reactive paradigms

What is reactive programming?

Reactive programming is a programming paradigm, or model, that centers around the concept of reacting
to changes in data and events as opposed to waiting for an event to happen.

An event in computing refers to user or system actions that trigger a specific response within a program.
These events are best visualized as "streams" that can flow through multiple processing elements, be
stopped along the way or fork into a parallel processing activity. For most cases, this processing is time-
sensitive, meaning that the applications require a different programming style, which is how reactive
programming came about.

Reactive programming relies on asynchronous programming logic to handle real-time updates to otherwise
static content. This means that a program can start a long-running task while, at the same time, still being
responsive to other events.

Reactive programming lets developers handle asynchronous data streams and events in a more intuitive
way by simplifying otherwise complex event-driven scenarios. It, in essence, is a way to
create software that responds to events rather than soliciting inputs from users. Using this programming
model, developers can create responsive and efficient event-driven applications.

Early applications of reactive programming to business applications were largely confined to things such as
monitoring the state of networks, servers, and software as well as signaling database conditions, such as
inventory levels. This focus is changing with the advent of the internet of things (IoT), smart buildings
and cities, and public cloud computing.

IoT has made the reactive model important in facilities management, industrial process control and even
home automation. The cloud has introduced both a style of componentizing software -- functional
computing and microservices -- and a movement to shift many reactive applications to the cloud for its
scalability and reliability benefits.

How does reactive programming work?

Reactive programming is all about streams, which are time-ordered sequences of related event messages.
Data streams used in reactive programming are created on a continual or near-continual basis. These data
streams are sent from a source -- a motion sensor, temperature gauge or product inventory database -- in
reaction to a trigger.

That trigger could be any of the following:

 An event, such as software-generated alerts, keystrokes or signals from an IoT system.
 A call, which is a function that invokes a routine as part of a workflow.
 A message, which is an information unit that the system sends back to the user or system operator
with information about the status of an operation, an error, failure or other condition.
Any asynchronous operations can also act as a trigger.
Reactive programming and the reactive systems it deals with also consist of a combination of observer and
handler functions. The observer function recognizes important conditions or changes, and then generates
messages to signal they've happened. The event handler is what deals with those messages appropriately.
A given stream will generally start with an observer. This can be either a segment of code in an application
that watches for a condition related to the application or a device such as an IoT sensor that generates an
event. A stream is sometimes diagrammed as an arrow -- left to right -- that starts with the observer
process and flows through one or more handlers until it's completely processed, terminates in an error
status or forks into derivative streams. Reactive programming is about building those observers and
handlers as well as threading the stream as required.

Device-generated streams are easily understood. But streams generated by software-inserted observers
are a bit more complicated. Normally, these elements work either in cooperation with the processing work
done by an application or they run periodically to monitor a database element. When this software element
recognizes a condition, it generates an event in the stream.

An event stream is steered either by the handlers themselves where work is dispatched to a specific next
process, or by a message bus such as an enterprise service bus or message queue that carries the
message to designated bus listeners. The message handling process determines whether a message is
broadcast to multiple handlers or to a single handler. It is also normally responsible for load-
balancing among multiple parallel handlers or providing spare handlers in the case of a failure.

[Link]

Each handler must either pass the message along, determine that the stream process has ended and "eat"
the message, or generate an error. The handler may decide whether to fork a message to multiple streams
or to generate a new stream. These fork conditions are often used to separate tasks in message handling;
for example, a message might generate a local response to open a gate as well as a message to a
transaction processing system.

The presumption in reactive programming is that there's no control over the number or timing of the events,
so the software must be resilient and highly scalable to manage variable loads. Instead of writing more
simple sequential code, developers must write everything as callback functions.

In "The Reactive Principle," the follow-up to "The Reactive Manifesto," Jonas Bonér et al. defines the eight
principles an application must embody to be considered reactive:

 Stay responsive. Always respond in a timely manner.

 Accept uncertainty. Build reliability despite unreliable foundations.
 Embrace failure. Expect things to go wrong and build for resilience.
 Assert autonomy. Design components that act independently and interact collaboratively.
 Tailor consistency. Individualize consistency per component to balance availability and
performance.
 Decouple time. Process asynchronously to avoid coordination and waiting.
 Decouple space. Create flexibility by embracing the network.
 Handle dynamics. Continuously adapt to varying demands and resources.
The Reactive Principles refer to eight different principles that an application must consider when it comes to
being reactive.

Benefits and challenges of reactive programming

The primary benefits of reactive programming techniques are their ability to do the following:
 Provide better control over the response times associated with the processing of events.
 Enable consistency in software design for real-time systems to reduce development, maintenance
costs, and effort.
 Support load balancing and resiliency to improve the quality of experience.
 Make the concept of a stream or event flow explicit, improving overall management of compute
elements and processing resources by making them more visual.
 Handle asynchronous and non-asynchronous code. This provides developers with the flexibility to
write code that can handle multiple tasks at the same time without having to cease a main thread
from executing.

These benefits come with challenges, however:

 Adding observer processes to current software might be difficult or impossible depending
on source code availability and staff programming skills.
 Reactive design is a major mindset shift for developers, and efforts will present a learning curve
during which more validation and supervision of design and coding might be required.
 Reactive systems can easily accumulate delay through an excessive number of processes linked
to the stream.
 Debugging an application that uses reactive programming is more complex due to the code’s
event-driven and asynchronous nature.

Still, modern web apps and mobile apps can be highly interactive, making use of many data events.
Reactive programming is a way to make these applications respond in real time in a scalable manner.
[Link]
Adopting reactive programming

There are several different approaches reactive programming can take. For example, reactive
programming can be integrated with imperative, actor-based, rule-based or object-oriented programming.
Developers will also have to choose the reactive programming framework. For example, reactive
programming in Java includes frameworks such as the following:
 RxJava. This is a functional reactive programming library used for Android.
 Akka. This is a toolkit and runtime for building applications on the Java Virtual Machine.
 Vert.x. This is another toolkit for building reactive applications on the Java Virtual Machine and is
structured as an event-driven and non-blocking architecture.
 Spring Framework 5.0. The Spring Framework is a reactive programming framework that uses
Reactive Streams to communicate between libraries and any asynchronous elements.

The Java Virtual Machine interprets bytecode and converts it to machine language that is platform specific.

Good reactive programs start with a clear diagram of the event stream, one that includes all the specific handler
processes and their role in processing, terminating or error generation. When this is done, the hosting platform --
edge, cloud or data center -- is selected and designated on the stream diagram for each process, avoiding any back
and forth across hosting platform boundaries. Development can then begin.

The following best practices should be observed while building reactive programming applications:

 Where an event stream must trigger a real-world response, such as opening a gate, keep
the control loop short by moving the responding process closer to the front of the stream and
hosting it near the event source.
 Avoid using programming languages and techniques that create stateful components that store
data with the software to ensure that components can be scaled and replaced easily.
 Review the location and implementation of any databases needed by any of the handler processes
to ensure that database access doesn't add latency or cross-cloud boundaries, generating
additional costs.
  At every step in development, reference the work done back to the event stream diagram to ensure
it's maintained, up-to-date and accurate.

Important Considerations for Reactive Programming

Problem: We want to create an application that displays real-time stock price updates for a specific ticker
symbol. Traditional polling methods would constantly fetch data from an API, even if there are no changes.
This can be inefficient and consume unnecessary resources.

Reactive Solution:
1. Data Stream: Establish a reactive stream that emits stock price updates from the API. This could
be implemented using a library like Project Reactor or RxJava.

2. Subscription: Subscribe to the data stream in your application. This means your code will be
notified whenever a new stock price update is emitted. Hot Observables emit data regardless of
whether there are subscribers. They are like a radio broadcast, always transmitting. Once
subscribed, a subscriber receives all emitted items, including those emitted before the subscription.
Cold Observables only emit data when subscribed. They are like a video on demand, only playing
when requested. Each subscriber gets their own independent sequence of emissions. Choose hot
Observables for shared data streams like stock prices or sensor readings. Choose cold
Observables when each subscriber needs their own personalized sequence, like user interactions
or search results. Hot Observables can lead to unexpected behavior if not managed carefully, while
cold Observables ensure predictable and controlled data flow.

3. Processing: When a new update arrives, process it and update the UI to display the latest price.

4. Error Handling: Implement error handling mechanisms to deal with potential exceptions, such as
network failures or API errors. Error handling in reactive streams is crucial to prevent unexpected
behavior and ensure system resilience. Strategies include:
 Error propagation: Allow errors to propagate downstream, allowing consumers to handle them
appropriately.
 Error recovery: Use operators like retry, retryWhen, or onErrorResumeNext to attempt retries or
switch to alternative data sources.
 Error termination: Use onErrorReturn, onErrorReturnItem, or onErrorComplete to terminate the
stream with a default value or completion.
 Error transformation: Use onErrorMap to transform errors into a different type for easier handling.
 Error isolation: Use onErrorContinue to ignore specific errors and continue processing.
 Error logging: Use onError to log errors for debugging and monitoring. By carefully considering
these strategies, you can effectively manage errors in reactive streams, preventing indefinite
propagation and ensuring graceful recovery.
5. Backpressure: If the rate of updates is too high, implement backpressure to control the flow of data and
prevent overwhelming the application. Backpressure is a mechanism in reactive programming that allows
downstream consumers to signal upstream producers when they are overwhelmed or cannot keep up with
the rate of data production. This prevents resource exhaustion by avoiding situations where producers
generate data faster than consumers can process it, leading to memory overflow or performance
degradation. By implementing backpressure, consumers can effectively control the flow of data, ensuring
that producers only generate data at a rate that can be consumed, thereby maintaining a balanced and
efficient system.
This reactive approach ensures that the application is only notified when there are actual changes in the
stock price, improving efficiency and responsiveness.

import [Link];
import [Link];
import [Link];

import [Link];
import [Link];

import [Link];
import [Link];
public class StockPriceUpdater {
private final WebClient webClient;
private final [Link]<StockPrice> priceSink;
private final AtomicReference<StockPrice> latestPrice;

public StockPriceUpdater(String apiBaseUrl, String tickerSymbol) {

[Link] = [Link]()
.baseUrl(apiBaseUrl)
.build();
[Link] = [Link]().multicast().onBackpressureBuffer();
[Link] = new AtomicReference<>();

// Create a flux that emits stock price updates

Flux<StockPrice> priceFlux = [Link]([Link](1))
.flatMap(i -> fetchStockPrice(tickerSymbol))
.doOnError(ex -> [Link]("Error fetching stock price: " + [Link]()))
.retryBackoff(3, [Link](1), [Link](10))
.subscribeOn([Link]());

// Subscribe to the flux and update the UI

[Link](price -> {
[Link](price);
[Link](price);
updateUI(price);
});
}

private StockPrice fetchStockPrice(String tickerSymbol) {

return [Link]()
.uri("/stock/{tickerSymbol}", tickerSymbol)
.retrieve()
.bodyToMono([Link])
.block();
}

private void updateUI(StockPrice price) {

// Update your UI component with the latest price
[Link]("Updated stock price: " + [Link]() + " - " + [Link]());
}

public Flux<StockPrice> getPriceFlux() {

return [Link]();
}

public static class StockPrice {

private String symbol;
private double price;

// Getters and setters

 public String getSymbol() {
return symbol;
}

public void setSymbol(String symbol) {

[Link] = symbol;
}

public double getPrice() {

return price;
}

public void setPrice(double price) {

[Link] = price;
}
}
}

public class Main {

public static void main(String[] args) {
String apiBaseUrl = "[Link]
String tickerSymbol = "AAPL";

StockPriceUpdater updater = new StockPriceUpdater(apiBaseUrl, tickerSymbol);

// Subscribe to the price flux

[Link]()
.subscribeOn([Link]())
.subscribe(price -> [Link]("Received updated price: " + [Link]() + " - " +
[Link]()));

[Link]

When to Use Reactive Programming

 I/O-bound applications: When your application heavily relies on network or disk operations.
 Event-driven systems: For applications that need to react to continuous streams of events.
 High-concurrency scenarios: When you need to handle many concurrent requests efficiently.
 Real-time applications: For systems that require low-latency responses.

Reactive programming is particularly well-suited for I/O-bound applications due to its asynchronous
nature. Unlike traditional blocking I/O, reactive programming allows the application to continue processing
other tasks while waiting for I/O operations to complete. This non-blocking approach significantly improves
scalability and responsiveness, especially in scenarios where I/O operations can be time-consuming. By
leveraging reactive streams and operators, developers can effectively manage asynchronous workflows,
ensuring that the application remains responsive and efficient even under heavy I/O loads.

Event-driven systems rely on asynchronous communication to handle continuous streams of events.

Reactive programming provides a natural framework for building such systems. By representing events as
Observables, developers can easily compose and transform event streams using reactive operators. This
enables the creation of complex event processing pipelines, where events can be filtered, mapped,
aggregated, and reacted to in a declarative manner. Additionally, reactive programming’s backpressure
mechanism helps manage the flow of events, preventing overload and ensuring that the system can handle
incoming events efficiently.

High-concurrency scenarios often pose significant challenges for traditional programming models.
Reactive programming, with its non-blocking and asynchronous nature, is well-equipped to handle such
workloads. By avoiding blocking operations, reactive applications can efficiently manage concurrent
requests, preventing resource contention and ensuring responsiveness. Additionally, reactive
programming’s backpressure mechanism helps regulate the flow of requests, preventing overload and
ensuring that the system can handle incoming requests in a controlled manner. This makes reactive
programming an ideal choice for applications that need to handle a large number of concurrent users or
requests.

Real-time applications require low-latency responses and the ability to process data in a timely manner.
Reactive programming’s asynchronous and non-blocking nature aligns well with the requirements of real-
time systems. By avoiding blocking operations and leveraging non-blocking I/O, reactive applications can
minimize latency and ensure that data is processed efficiently. Additionally, reactive programming’s event-
driven model allows for rapid responses to incoming data, making it suitable for applications that need to
react quickly to changes in their environment.

What are reactive streams?

Reactive Streams is a specification founded in 2013. It defines the interfaces and classes that are used to
create applications through reactive programming. The aim of classes Reactive Streams is to specify
interfaces, protocols and frameworks for reactive programming while also defining a standard for
asynchronous stream processing with non-blocking backpressure.
Asynchronous stream processing with non-blocking backpressure refers to the processing of data streams
asynchronously, meaning that receiving systems can handle the flow of data without getting overwhelmed.
Libraries and frameworks such as RxJava, Akka and Spring Framework 5.0 were created using the
Reactive Streams specification, for example.
Reactive programming use cases
The primary use cases for reactive programming include the following:
 IoT applications where sensors create events that then control real-world process steps, create
business transactions or both. This is the fastest-growing application of reactive programming
techniques, though not the traditional target.
 Applications that gather status information from networks or data processing elements through
inserted software agents to monitor activities or data elements. This is the first classic reactive
programming application, but one converging with IoT.
 Any application that requires highly interactive user-to-user interface handling, especially where
each keystroke must be processed and interpreted. This is the other classic reactive programming
application, and it now includes gaming, web apps and some social media applications.
 Signaling between applications, particularly between what could be called foreground applications
and background applications, that perform statistical analysis and database cleanup. This use case
will normally involve a daemon process that monitors for changes and activates an event stream
when one is detected.
 Coordination between functional AWS Lambda cloud processing and back-end data center
processing, where an event will trigger the execution of a back-end process. This facilitates some
forms of hybrid cloud development.
  Real-time data streaming and big data analysis. Reactive programming is ideal for handling large
amounts of real-time streaming data.

 Observer pattern and pub-sub models

The Observer Pattern and Publish-Subscribe (Pub-Sub) models are both design patterns used to
handle communication between components in a system — particularly when one part needs to react to
changes or events occurring in another part — but they differ in how they achieve this communication.
Let’s break them down:

🔹 Observer Pattern
Concept:
The Observer Pattern defines a one-to-many dependency between objects, so that when one object (the
subject) changes state, all its dependents (observers) are notified automatically.
Key idea:
Direct communication — the subject keeps references to its observers and notifies them directly.
Structure:
 Subject (Observable): Maintains a list of observers and provides methods to attach/detach them.
 Observers: Implement an update interface to receive notifications from the subject.
Example (Conceptual):
// Subject
class WeatherStation {
private List<Observer> observers = new ArrayList<>();
private float temperature;

public void addObserver(Observer o) { [Link](o); }

public void removeObserver(Observer o) { [Link](o); }

public void setTemperature(float temp) {

[Link] = temp;
notifyObservers();
}

private void notifyObservers() {

for (Observer o : observers) {
[Link](temperature);
}
}
}

// Observer interface
interface Observer {
void update(float temperature);
}

// Concrete Observer
class Display implements Observer {
public void update(float temperature) {
[Link]("New temperature: " + temperature);
 }
}
Use case examples:
 GUI frameworks (e.g., buttons and event listeners)
 Model-View-Controller (MVC) architecture
 Real-time data updates (e.g., dashboards)

🔹 Publish-Subscribe (Pub-Sub) Model

Concept:
The Publish-Subscribe model is similar in spirit to the observer pattern but introduces an intermediary
component — a message broker — to decouple publishers from subscribers.
Key idea:
Indirect communication — publishers send messages to a channel/topic; subscribers listen to that topic,
and the broker delivers messages.
Structure:
 Publisher: Sends (publishes) messages to a topic or channel.
 Subscriber: Subscribes to specific topics of interest.
 Broker (or Event Bus): Mediates between publishers and subscribers, managing message
delivery.
Example (Conceptual):
// Using a simple message broker
class EventBroker {
private Map<String, List<Consumer<String>>> subscribers = new HashMap<>();

public void subscribe(String topic, Consumer<String> handler) {

[Link](topic, k -> new ArrayList<>()).add(handler);
}

public void publish(String topic, String message) {

if ([Link](topic)) {
for (Consumer<String> handler : [Link](topic)) {
[Link](message);
}
}
}
}

// Usage
EventBroker broker = new EventBroker();

[Link]("news", msg -> [Link]("Subscriber 1 received: " + msg));

[Link]("news", msg -> [Link]("Subscriber 2 received: " + msg));

[Link]("news", "Breaking News: Cloud systems are scaling!");

Use case examples:
 Event-driven architectures
 Messaging systems (e.g., Kafka, RabbitMQ, MQTT)
 Microservices communication
 Cloud notification systems
🔸 Comparison Table
Feature Observer Pattern Publish-Subscribe Model
Tight coupling (observers know the
Coupling Loose coupling (via broker or event bus)
subject)
Indirect (publisher → broker →
Communication Direct (subject → observer)
subscriber)
Scalability Limited to in-process communication Scales across distributed systems
Example
Java Observer/Observable, GUI listeners MQTT, Apache Kafka, Google Pub/Sub
Frameworks
Use Case Real-time UI updates, MVC Event-driven systems, distributed apps

✅ Summary
 Use the Observer Pattern when you need direct, in-memory updates between objects in the same
application (e.g., GUI, MVC).
 Use the Pub-Sub Model when you need asynchronous, decoupled communication across
different systems or services (e.g., distributed or cloud-based systems).

 Debouncing and throttling events

⚙️Debouncing and Throttling Events

When dealing with high-frequency events — like scrolling, resizing, keypresses, or mouse movements
— your application might receive hundreds or thousands of event calls per second.
Processing all these events can lead to:
 High CPU/memory usage
 UI freezing or input lag
 Unnecessary function calls
To handle this efficiently, developers use debouncing and throttling.

🧩 1. Debouncing
🔹 Definition
Debouncing ensures that a function is executed only after a certain period of inactivity has elapsed
since the last event.
In other words, the function runs after the event stops firing.
🔹 Purpose
To limit how often a function is executed by waiting until rapid events have finished.

🧠 Concept
 Every time the event fires, a timer is reset.
 If another event happens before the timer expires, the timer restarts.
 The function executes only once — after events stop firing for a set delay.

📘 Example Use Cases

 User stops typing → perform a search query (e.g., autocomplete box)
 Window resizing → re-render layout only after resizing ends
  Button clicks → prevent accidental double submissions

💻 Example (JavaScript)
function debounce(func, delay) {
let timer;
return function(...args) {
clearTimeout(timer);
timer = setTimeout(() => [Link](this, args), delay);
};
}

// Example: Debounced search input

const handleSearch = debounce((query) => {
[Link]("Searching for:", query);
}, 500);

[Link]("searchBox").addEventListener("input", (event) => {

handleSearch([Link]);
});
Explanation:
 debounce() delays execution of handleSearch until the user stops typing for 500 ms.
 If the user keeps typing quickly, the timer resets each time.
 The function runs only once after the last keystroke.

✅ Advantages
 Prevents redundant or excessive calls.
 Improves performance on rapid-fire events.
 Ideal for input fields, resize, and scroll-based events.
⚠️Disadvantages
 Adds a slight delay to the function’s execution.
 Not suitable for tasks needing immediate reaction (e.g., real-time tracking).

⏱ 2. Throttling
🔹 Definition
Throttling ensures that a function executes at regular intervals during continuous event firing.
It limits a function to run at most once per specified time frame, no matter how many times the event is
triggered.

🧠 Concept
 The first event triggers the function immediately.
 Then, the function is blocked for a defined period (e.g., 200ms).
 After the interval, the function can be executed again if new events occur.

📘 Example Use Cases

 Tracking scroll position during scrolling
 Monitoring mouse movement (e.g., drag or draw)
 Handling window resizing updates smoothly

💻 Example (JavaScript)
function throttle(func, limit) {
let inThrottle;
return function(...args) {
if (!inThrottle) {
[Link](this, args);
inThrottle = true;
setTimeout(() => (inThrottle = false), limit);
}
};
}

// Example: Throttled scroll event

const handleScroll = throttle(() => {
[Link]("Scroll event triggered");
}, 1000);

[Link]("scroll", handleScroll);

Explanation:
 The function handleScroll() runs at most once per second, no matter how often the user scrolls.
 This keeps performance stable, especially for heavy computations.

✅ Advantages
 Controls function call rate.
 Ideal for continuous events (scroll, resize, mousemove).
 Ensures responsiveness while preventing overload.
⚠️Disadvantages
 Some events between intervals are ignored.
 May not capture the final event state immediately.

⚖️3. Comparison: Debounce vs Throttle

Aspect Debouncing Throttling
Executes function after a specified Executes function at regular intervals during
Definition
period of inactivity. continuous activity.
Execution Timing After event ends. During event, at set intervals.
Group many rapid events into one Ensure steady execution rate during
Purpose
execution. continuous events.
Typical Delay
Waits until the last event. Executes immediately, then waits.
Behavior
Search input, resize end, form Scroll tracking, drag-and-drop, mouse
Use Cases
validation. movement.
Example Delay Function runs after 500ms of no Function runs every 500ms while events
(500ms) events. occur.
Performance Focus Reduces redundant calls. Controls execution frequency.
Result Behavior Fires once after user stops. Fires periodically while user continues.

🧠 4. Practical Scenarios
Scenario Technique Why
User typing in a search bar Debounce Wait until user stops typing before fetching results.
Resizing browser window Debounce Wait until resize finishes before recalculating layout.
Update UI (like showing scroll progress) at intervals, not every
Scrolling page Throttle
pixel.
Mouse drag tracking Throttle Limit updates to avoid performance drops.
Button clicks Debounce Prevent accidental double submission.

🧮 5. Real-World Applications
Domain Example Technique
Web Apps Autocomplete suggestions Debounce
IoT Systems Sensor data sampling Throttle
Mobile Apps Gesture tracking Throttle
Backend APIs Rate-limiting requests Throttle
Form Validation Delay validation until user stops typing Debounce

🚀 6. Summary
 Debouncing:
→ Executes once after events stop firing.
→ Best for “after-the-fact” actions.
 Throttling:
→ Executes at fixed intervals during event bursts.
→ Best for “continuous monitoring” actions.

Both techniques are used to enhance performance, reduce CPU usage, and improve user experience
by controlling event-driven function executions.
⚙️Debouncing and Throttling in Java
Both concepts revolve around controlling how often a function executes when triggered by rapid or
repeated events — such as sensor updates, UI inputs, or network requests.

🧩 1. Debouncing in Java
🔹 Definition
Debouncing ensures that a method runs only after a specified period has passed since the last event.
If another event occurs before that delay expires, the timer resets.
In other words, the function executes only once after the "noise" stops.

💡 Conceptual Use Case

Imagine you have a text field for searching.
You don’t want to send a database query for every keystroke — only after the user stops typing for a short
time (e.g., 500 ms).

💻 Example: Debounce Implementation in Java

import [Link];
import [Link];

public class Debouncer {

private final long delay;
private Timer timer;

public Debouncer(long delay) {

[Link] = delay;
[Link] = new Timer();
}

public void call(Runnable task) {

// Cancel previous scheduled task
if (timer != null) {
[Link]();
}
timer = new Timer();

// Schedule new one after delay

[Link](new TimerTask() {
@Override
public void run() {
[Link]();
}
}, delay);
}
 public static void main(String[] args) throws InterruptedException {
Debouncer debouncer = new Debouncer(1000); // 1 second debounce

// Simulate rapid input events

for (int i = 0; i < 5; i++) {
[Link](() -> [Link]("Search executed at: " + [Link]()));
[Link](300); // 300ms between keystrokes
}

// Wait to allow final call to execute

[Link](2000);
}
}
🧠 Explanation
 Each keystroke cancels the previous timer.
 The Runnable executes only after 1 second of no new events.
 Output shows only one final execution, even though multiple calls were made.
✅ Use Cases
 Search boxes (run query after typing stops)
 Button debounce (prevent double-clicks)
 Network retry mechanisms (delay requests after burst)

⏱ 2. Throttling in Java
🔹 Definition
Throttling ensures that a method executes at most once every specified interval, no matter how
frequently events occur.
It doesn’t wait for inactivity — it runs at a controlled rate.

💡 Conceptual Use Case

If you’re monitoring a sensor that emits hundreds of readings per second, you might only need to process
one reading per second to save CPU and power.

💻 Example: Throttle Implementation in Java

public class Throttler {
private final long interval;
private long lastExecutionTime = 0;

public Throttler(long interval) {

[Link] = interval;
}

public void call(Runnable task) {

long now = [Link]();
if (now - lastExecutionTime >= interval) {
lastExecutionTime = now;
[Link]();
}
}
 public static void main(String[] args) throws InterruptedException {
Throttler throttler = new Throttler(1000); // 1 second throttle

// Simulate rapid event stream

for (int i = 0; i < 10; i++) {
[Link](() -> [Link]("Event handled at: " + [Link]()));
[Link](200); // Event fires every 200ms
}
}
}
🧠 Explanation
 The function runs immediately the first time.
 It ignores events until at least 1 second passes since the last run.
 This ensures a steady, periodic execution rate.
✅ Use Cases
 Sensor data sampling (IoT systems)
 API rate limiting
 Scroll or resize event handling (Swing/JavaFX)
 Logging systems (avoid flooding logs)

⚖️3. Comparison: Debounce vs Throttle in Java

Aspect Debouncing Throttling
Execution Timing After events stop firing At fixed time intervals
Delay Behavior Waits for a pause before executing Executes regularly during event stream
Consolidate many events into one final
Focus Limit call rate during continuous events
call
[Link]() check or
Implementation Tool Timer, ScheduledExecutorService
scheduler
Ideal For Text input, validation, search, resizing Sensor data, continuous input, API limits
Example Delay (1000 Executes once after 1 second of Executes every 1 second if events
ms) inactivity continue

⚙️4. Advanced Version Using ExecutorService (Thread-Safe)

Using ScheduledExecutorService is better for multi-threaded or server environments.
💻 Debounce (Thread-Safe Version)
import [Link].*;

public class DebouncerSafe {

private final ScheduledExecutorService scheduler = [Link]();
private ScheduledFuture<?> future;
private final long delay;

public DebouncerSafe(long delay) {

[Link] = delay;
}

public synchronized void call(Runnable task) {

if (future != null && ![Link]()) {
 [Link](false);
}
future = [Link](task, delay, [Link]);
}

public void shutdown() {

[Link]();
}

public static void main(String[] args) throws InterruptedException {

DebouncerSafe debouncer = new DebouncerSafe(1000);

for (int i = 0; i < 5; i++) {

[Link](() -> [Link]("Debounced at: " + [Link]()));
[Link](250);
}

[Link](2000);
[Link]();
}
}

💻 Throttle (Thread-Safe Version)

import [Link].*;

public class ThrottlerSafe {

private final long interval;
private long lastExecution = 0;
private final Object lock = new Object();

public ThrottlerSafe(long interval) {

[Link] = interval;
}

public void call(Runnable task) {

long now = [Link]();
synchronized (lock) {
if (now - lastExecution >= interval) {
lastExecution = now;
[Link]().execute(task);
}
}
}

public static void main(String[] args) throws InterruptedException {

ThrottlerSafe throttler = new ThrottlerSafe(1000);

for (int i = 0; i < 10; i++) {

[Link](() -> [Link]("Throttled call at: " + [Link]()));
[Link](200);
 }
}
}

🧠 5. Summary
Aspect Debouncing Throttling
Execute function once after burst of events Execute function at controlled
Goal
ends intervals
Execution Timing After inactivity Periodic during activity
Ignores new events until interval
Delay Reset Resets timer each event
passes
Example Use Case Search box, resize, form validation Sensor updates, logging, rate limiting
Common Tools in
Timer, ScheduledExecutorService Time checks, ExecutorService
Java

✅ In short:
 Use debouncing when you want to wait for silence (e.g., user finished typing).
 Use throttling when you want to control frequency (e.g., limit event rate).

 Event streams and pipelines

Event Streams and Event Pipelines, two fundamental concepts in event-driven architecture (EDA), IoT
systems, and real-time data processing (like Kafka, Flink, or reactive programming in Java).

⚙️Event Streams and Pipelines

🔹 Overview
Modern systems are increasingly event-driven — they respond to data as it happens, rather than in
periodic batches.
In such systems, event streams represent a continuous flow of event data, while event pipelines define
how those events are processed, transformed, and routed through various stages.
Together, they enable real-time analytics, IoT telemetry, log aggregation, and microservice
coordination.
🧩 1. Event Streams
🔹 Definition
An event stream is a continuous, ordered sequence of event data records that represent actions,
changes, or updates occurring in a system over time.
Each event represents a state change — something that happened — and contains:
 Timestamp (when it happened)
 Event type (what happened)
 Payload (contextual data)
 Metadata (source, ID, etc.)
Think of an event stream as an append-only log where events are constantly added and consumed in real
time.

💡 Examples of Event Streams

Domain Event Examples
IoT Systems Sensor readings: temperature, humidity, motion
E-commerce Order placed, payment completed, item shipped
Banking Transaction processed, balance updated
Social Media User posted, liked, or commented
DevOps Log entries, system metrics, server alerts

💻 Conceptual Example in Java

Let’s define a simple event stream structure.
class Event {
private final String type;
private final long timestamp;
private final String payload;

public Event(String type, String payload) {

[Link] = type;
[Link] = payload;
[Link] = [Link]();
}

public String getType() { return type; }

public long getTimestamp() { return timestamp; }
public String getPayload() { return payload; }

@Override
public String toString() {
return "[" + timestamp + "] " + type + " => " + payload;
}
}
We can simulate a continuous event stream (e.g., sensor data):
import [Link].*;
import [Link].*;

public class EventStreamDemo {

 public static void main(String[] args) {
ScheduledExecutorService executor = [Link](1);

Runnable generateEvent = () -> {

Event event = new Event("TemperatureReading", "Value=" + new Random().nextInt(40));
[Link]("Produced: " + event);
};

// Emit events every second

[Link](generateEvent, 0, 1, [Link]);
}
}
Output (Sample):
Produced: [1730191200000] TemperatureReading => Value=29
Produced: [1730191201000] TemperatureReading => Value=31
Produced: [1730191202000] TemperatureReading => Value=33

🧠 Characteristics of Event Streams

Property Description
Continuous Events occur indefinitely, not in batches.
Immutable Events are append-only (cannot be modified).
Ordered Events are time-sequenced and consumed in order.
Asynchronous Producers and consumers operate independently.
Scalable Streams can be partitioned for high throughput.

⚙️2. Event Pipelines

🔹 Definition
An event pipeline defines the pathway through which events move — from production to processing to
storage or action.
It’s the processing workflow that transforms raw events into meaningful insights or triggers.
In essence:
Event Streams are what happens,
Event Pipelines define how we handle it.

🔹 Typical Stages of an Event Pipeline

Stage Description Example Technologies
Events are generated by sensors, apps, or IoT devices, Kafka producers, web
Event Production
microservices. apps
Events enter the system via a broker or Apache Kafka, RabbitMQ, MQTT,
Event Ingestion
queue. AWS Kinesis
Event Stream Events are filtered, transformed, aggregated, Apache Flink, Spark Streaming,
Processing or enriched. Akka Streams
Storage and Analytics Events are persisted for historical analysis. Cassandra, ElasticSearch, HDFS
Other services or applications respond to Microservices, alert systems,
Action/Reaction
processed events. dashboards
💻 Example: Event Pipeline in Java (Simplified)
import [Link].*;
import [Link].*;

class EventPipeline {
private final BlockingQueue<Event> queue = new LinkedBlockingQueue<>();

// Producer: generates events

public void produce(Event event) {
[Link](event);
}

// Consumer: processes events

public void consume() {
[Link]().execute(() -> {
try {
while (true) {
Event event = [Link]();
processEvent(event);
}
} catch (InterruptedException e) {
[Link]().interrupt();
}
});
}

// Example transformation stage

private void processEvent(Event event) {
[Link]("Processing: " + event);
if ([Link]().equals("TemperatureReading")) {
int value = [Link]([Link]().split("=")[1]);
if (value > 35) {
[Link]("⚠️ALERT: High temperature detected: " + value);
}
}
}
}

public class EventPipelineDemo {

public static void main(String[] args) {
EventPipeline pipeline = new EventPipeline();
[Link]();

ScheduledExecutorService executor = [Link](1);

Runnable producerTask = () -> [Link](
new Event("TemperatureReading", "Value=" + new Random().nextInt(40))
);

[Link](producerTask, 0, 1, [Link]);
}
}
Output:
Processing: [1730191200000] TemperatureReading => Value=31
Processing: [1730191201000] TemperatureReading => Value=38
⚠️ALERT: High temperature detected: 38
Processing: [1730191202000] TemperatureReading => Value=36
⚠️ALERT: High temperature detected: 36

🧠 How Pipelines Work

1. Ingest: Collect raw event data (e.g., sensor output, API logs).
2. Transform: Enrich, clean, or filter data (e.g., add metadata, remove duplicates).
3. Analyze: Perform computations or pattern detection (e.g., moving averages, anomaly detection).
4. Output: Store results, trigger alerts, or call other services.

☁️3. Event Streams and Pipelines in Distributed Systems

Component Role
Producers Send events (e.g., IoT devices, web services).
Message Broker Buffers and routes events (Kafka, MQTT).
Stream Processors Analyze, aggregate, or enrich event data (Flink, Spark Streaming).
Consumers Store or react to processed events (databases, dashboards, microservices).
Typical Flow:
IoT Device → Kafka Broker → Flink Stream Processor → Database → Dashboard/Alert

🧮 4. Advantages of Event Streams & Pipelines

Benefit Explanation
Real-time Processing Reacts instantly to new data (no delay).
Decoupled Architecture Producers and consumers operate independently.
Scalability Handle millions of events per second via partitioning.
Fault Tolerance Systems can recover from crashes via durable logs.
Flexibility Easy to add new processing stages or consumers.

⚠️5. Challenges
Challenge Description
Event Ordering Maintaining correct order in distributed systems.
Data Duplication Handling repeated events or retries.
Fault Handling Ensuring “at-least-once” or “exactly-once” delivery.
Latency Management Balancing speed with processing accuracy.
Schema Evolution Managing changing event formats over time.

🔄 6. Real-World Examples
Platform Purpose
Apache Kafka Distributed event streaming platform used in enterprise systems.
Apache Flink / Spark Streaming Real-time event processing and analytics.
AWS Kinesis / Google Pub/Sub Managed cloud-based event streaming.
Reactive Streams (Java 9+) Asynchronous event handling with backpressure support.
Platform Purpose
Spring Cloud Stream Simplified event pipelines for microservices in Spring Boot.

🧠 7. Event Streams in Java Ecosystem

🔹 Reactive Streams API (Java 9+)
The Reactive Streams specification provides a Publisher–Subscriber model for event pipelines, handling
asynchronous streams of data with backpressure control.
import [Link].*;

class SimplePublisher implements Publisher<String> {

@Override
public void subscribe(Subscriber<? super String> subscriber) {
[Link](new Subscription() {
public void request(long n) {
for (int i = 1; i <= n; i++) {
[Link]("Event " + i);
}
[Link]();
}
public void cancel() {}
});
}
}

public class ReactiveStreamDemo {

public static void main(String[] args) {
Publisher<String> publisher = new SimplePublisher();
[Link](new Subscriber<>() {
public void onSubscribe(Subscription s) { [Link](5); }
public void onNext(String item) { [Link]("Received: " + item); }
public void onError(Throwable t) { [Link](); }
public void onComplete() { [Link]("Stream completed"); }
});
}
}

🧾 8. Summary Table
Aspect Event Stream Event Pipeline
Continuous flow of immutable event Processing workflow that handles and transforms
Definition
data events
Function Captures “what happens” in real time Defines “how to react” and “where to send” data
Nature Data source Data processing pathway
Persistence Often stored as logs (Kafka topics) Typically transient (processing chain)
Example in
Publisher<Stream<Event>> Processor<Event, ProcessedEvent>
Java
Focus Event generation and consumption Event transformation and routing
Use Cases IoT telemetry, app logs, clickstream Real-time analytics, alerts, ETL
✅ Summary
 Event Streams represent continuous data about system activities.
 Event Pipelines represent the flow of operations applied to those events.
 Together, they power real-time, scalable, event-driven systems.
 In Java, these concepts appear in frameworks like Reactive Streams, Spring Cloud Stream,
Kafka Streams, and Akka Streams.

Laboratory Exercises: Module 7 – Real-Time and Reactive Programming (Java, NetBeans)

🧪 Laboratory Exercise 1: Implementing the Observer Pattern

Objective
 To implement the Observer Pattern using Java.
 To understand the one-to-many dependency between subjects and observers.
Materials
 NetBeans IDE
 JDK 17 or higher
Procedure
1. Open NetBeans → Create a new Java Application project named ObserverPatternDemo.
2. Create three Java classes:
o WeatherStation (Subject)
o Observer (Interface)
o DisplayDevice (Concrete Observer)
3. Implement the code below.
Sample Code
import [Link];
import [Link];

interface Observer {
void update(float temperature);
}

class WeatherStation {
private List<Observer> observers = new ArrayList<>();
private float temperature;

public void addObserver(Observer o) { [Link](o); }

public void removeObserver(Observer o) { [Link](o); }

public void setTemperature(float temperature) {

[Link] = temperature;
notifyObservers();
}

private void notifyObservers() {

for (Observer o : observers) {
[Link](temperature);
}
}
}
class DisplayDevice implements Observer {
private String name;

public DisplayDevice(String name) {

[Link] = name;
}

@Override
public void update(float temperature) {
[Link](name + " displays new temperature: " + temperature + "°C");
}
}

public class ObserverPatternDemo {

public static void main(String[] args) {
WeatherStation station = new WeatherStation();

DisplayDevice phoneDisplay = new DisplayDevice("Phone Display");

DisplayDevice wallDisplay = new DisplayDevice("Wall Display");

[Link](phoneDisplay);
[Link](wallDisplay);

[Link](29.5f);
[Link](31.0f);
}
}
Expected Output
Phone Display displays new temperature: 29.5°C
Wall Display displays new temperature: 29.5°C
Phone Display displays new temperature: 31.0°C
Wall Display displays new temperature: 31.0°C

🧪 Laboratory Exercise 2: Publish–Subscribe Model Simulation

Objective
 To simulate the Pub-Sub model using a simple event broker.
 To show indirect communication between publishers and subscribers.
Procedure
1. Create a new Java project in NetBeans named PubSubModelDemo.
2. Create a class [Link] and copy the following code:
Sample Code
import [Link].*;
import [Link];

class EventBroker {
private Map<String, List<Consumer<String>>> subscribers = new HashMap<>();

public void subscribe(String topic, Consumer<String> handler) {

[Link](topic, k -> new ArrayList<>()).add(handler);
 }

public void publish(String topic, String message) {

if ([Link](topic)) {
for (Consumer<String> handler : [Link](topic)) {
[Link](message);
}
}
}
}

public class EventBrokerDemo {

public static void main(String[] args) {
EventBroker broker = new EventBroker();

[Link]("news", msg -> [Link]("Subscriber 1 received: " + msg));

[Link]("news", msg -> [Link]("Subscriber 2 received: " + msg));

[Link]("news", "Breaking News: Reactive programming is awesome!");

[Link]("news", "Update: Event-driven systems are now mainstream.");
}
}
Expected Output
Subscriber 1 received: Breaking News: Reactive programming is awesome!
Subscriber 2 received: Breaking News: Reactive programming is awesome!
Subscriber 1 received: Update: Event-driven systems are now mainstream.
Subscriber 2 received: Update: Event-driven systems are now mainstream.

🧪 Laboratory Exercise 3: Debouncing and Throttling Events

Objective
 To demonstrate Debouncing and Throttling mechanisms in Java event handling.
 To understand performance optimization during rapid event streams.
Procedure
1. Create a new Java project named DebounceThrottleDemo.
2. Create a class [Link] and paste this code.
Sample Code
import [Link];
import [Link];

class Debouncer {
private final long delay;
private Timer timer;

public Debouncer(long delay) { [Link] = delay; [Link] = new Timer(); }

public void call(Runnable task) {

if (timer != null) [Link]();
timer = new Timer();
[Link](new TimerTask() {
@Override
 public void run() { [Link](); }
}, delay);
}
}

class Throttler {
private final long interval;
private long lastExecutionTime = 0;

public Throttler(long interval) { [Link] = interval; }

public void call(Runnable task) {

long now = [Link]();
if (now - lastExecutionTime >= interval) {
lastExecutionTime = now;
[Link]();
}
}
}

public class DebounceThrottle {

public static void main(String[] args) throws InterruptedException {
[Link]("=== Debounce Example ===");
Debouncer debounce = new Debouncer(1000);

for (int i = 0; i < 5; i++) {

[Link](() -> [Link]("Debounced action executed at: " +
[Link]()));
[Link](250);
}

[Link](2000);

[Link]("\n=== Throttle Example ===");

Throttler throttler = new Throttler(1000);
for (int i = 0; i < 10; i++) {
[Link](() -> [Link]("Throttled action executed at: " + [Link]()));
[Link](200);
}
}
}
Expected Output (Example)
=== Debounce Example ===
Debounced action executed at: 1730200001000

=== Throttle Example ===

Throttled action executed at: 1730200002000
Throttled action executed at: 1730200003000
Throttled action executed at: 1730200004000
...
🧪 Laboratory Exercise 4: Event Streams and Pipelines
Objective
 To simulate event streaming and pipeline processing using Java threads and queues.
 To demonstrate producer–consumer flow in real time.
Procedure
1. Create a new Java project in NetBeans named EventPipelineDemo.
2. Add the following code to [Link].
Sample Code
import [Link].*;
import [Link].*;

class Event {
private final String type;
private final long timestamp;
private final String payload;

public Event(String type, String payload) {

[Link] = type;
[Link] = payload;
[Link] = [Link]();
}

public String getType() { return type; }

public String getPayload() { return payload; }

@Override
public String toString() {
return "[" + timestamp + "] " + type + " => " + payload;
}
}

class EventPipeline {
private final BlockingQueue<Event> queue = new LinkedBlockingQueue<>();

public void produce(Event event) { [Link](event); }

public void consume() {

[Link]().execute(() -> {
try {
while (true) {
Event event = [Link]();
process(event);
}
} catch (InterruptedException e) {
[Link]().interrupt();
}
});
}
 private void process(Event event) {
[Link]("Processing: " + event);
if ([Link]().equals("TemperatureReading")) {
int value = [Link]([Link]().split("=")[1]);
if (value > 35) {
[Link]("⚠️ALERT: High temperature detected: " + value);
}
}
}
}

public class EventPipelineDemo {

public static void main(String[] args) {
EventPipeline pipeline = new EventPipeline();
[Link]();

ScheduledExecutorService executor = [Link](1);

Runnable generateEvent = () -> {
Event event = new Event("TemperatureReading", "Value=" + new Random().nextInt(40));
[Link](event);
};

[Link](generateEvent, 0, 1, [Link]);
}
}
Expected Output
Processing: [1730201000000] TemperatureReading => Value=31
Processing: [1730201001000] TemperatureReading => Value=38
⚠️ALERT: High temperature detected: 38
Processing: [1730201002000] TemperatureReading => Value=36
⚠️ALERT: High temperature detected: 36

🧾 Submission Requirements
Each student should:
1. Create all four projects in NetBeans.
2. Run and test each program.
3. Capture screenshots of:
o Code editor view
o Output console view
4. Compile all screenshots and observations into a Lab Report (PDF or DOCX) with:
o Name and Section
o Objectives
o Procedure
o Output Screenshots
o Conclusion (2–3 sentences per activity)

💡 Instructor’s Note
These lab exercises reinforce:
 Event-driven and reactive design thinking.
 Java concurrency and timing control.
 Practical application of streams, observers, and pipelines in real-world systems (IoT, GUIs, APIs).