Java Observer Pattern — Missing removeObserver Causes OOM

import java.util.ArrayList;
import java.util.List;

// --- The Observer contract ---
// Any class that wants to receive updates must implement this.
interface StockObserver {
    void onPriceChanged(String tickerSymbol, double newPrice);
}

// --- The Subject ---
// Maintains a list of observers and notifies them on state change.
class StockMarket {

    private final List<StockObserver> observers = new ArrayList<>();
    private String tickerSymbol;
    private double currentPrice;

    public StockMarket(String tickerSymbol, double initialPrice) {
        this.tickerSymbol = tickerSymbol;
        this.currentPrice = initialPrice;
    }

    // Observers call this to sign up for notifications.
    public void addObserver(StockObserver observer) {
        observers.add(observer);
    }

    // Observers call this to unsubscribe — important to avoid memory leaks.
    public void removeObserver(StockObserver observer) {
        observers.remove(observer);
    }

    // Simulates a price update arriving from the exchange.
    public void updatePrice(double newPrice) {
        this.currentPrice = newPrice;
        notifyAllObservers(); // Automatically alerts every registered listener.
    }

    // Internal: loops through every registered observer and calls the contract method.
    private void notifyAllObservers() {
        for (StockObserver observer : observers) {
            observer.onPriceChanged(tickerSymbol, currentPrice);
        }
    }
}

// --- Concrete Observer 1: A trading dashboard ---
class TradingDashboard implements StockObserver {
    @Override
    public void onPriceChanged(String tickerSymbol, double newPrice) {
        // Each observer decides what to DO with the update independently.
        System.out.printf("[Dashboard] %s price updated on screen: $%.2f%n", tickerSymbol, newPrice);
    }
}

// --- Concrete Observer 2: An automated alert system ---
class PriceAlertSystem implements StockObserver {
    private final double alertThreshold;

    public PriceAlertSystem(double alertThreshold) {
        this.alertThreshold = alertThreshold;
    }

    @Override
    public void onPriceChanged(String tickerSymbol, double newPrice) {
        // This observer only acts when its own condition is met.
        if (newPrice > alertThreshold) {
            System.out.printf("[ALERT] %s crossed threshold! Current price: $%.2f%n", tickerSymbol, newPrice);
        }
    }
}

// --- Concrete Observer 3: An audit logger ---
class AuditLogger implements StockObserver {
    @Override
    public void onPriceChanged(String tickerSymbol, double newPrice) {
        System.out.printf("[Audit Log] Price change recorded — %s: $%.2f%n", tickerSymbol, newPrice);
    }
}

// --- Entry point ---
public class StockTicker {
    public static void main(String[] args) {
        StockMarket appleStock = new StockMarket("AAPL", 170.00);

        TradingDashboard dashboard = new TradingDashboard();
        PriceAlertSystem alertSystem = new PriceAlertSystem(185.00); // Alert fires above $185
        AuditLogger logger = new AuditLogger();

        // All three observers register with the same subject.
        appleStock.addObserver(dashboard);
        appleStock.addObserver(alertSystem);
        appleStock.addObserver(logger);

        System.out.println("--- Price update 1 ---");
        appleStock.updatePrice(180.50); // Below threshold — alert won't fire.

        System.out.println("--- Price update 2 ---");
        appleStock.updatePrice(186.00); // Above threshold — alert fires.

        System.out.println("--- Dashboard unsubscribes ---");
        appleStock.removeObserver(dashboard); // Dashboard opts out.

        System.out.println("--- Price update 3 ---");
        appleStock.updatePrice(190.00); // Dashboard no longer receives this.
    }
}

import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.beans.PropertyChangeSupport;

// --- The Subject using Java's built-in PropertyChangeSupport ---
// This is ideal for JavaBean-style domain objects.
class UserProfile {

    // PropertyChangeSupport does the heavy lifting of managing listeners.
    private final PropertyChangeSupport changeSupport = new PropertyChangeSupport(this);

    private String username;
    private String emailAddress;

    public UserProfile(String username, String emailAddress) {
        this.username = username;
        this.emailAddress = emailAddress;
    }

    // Standard registration method expected by the Java Beans convention.
    public void addPropertyChangeListener(PropertyChangeListener listener) {
        changeSupport.addPropertyChangeListener(listener);
    }

    public void removePropertyChangeListener(PropertyChangeListener listener) {
        changeSupport.removePropertyChangeListener(listener);
    }

    public void setEmailAddress(String newEmailAddress) {
        String previousEmail = this.emailAddress;
        this.emailAddress = newEmailAddress;
        // Fires the event with the property name, old value, AND new value.
        // Observers get context — not just 'something changed'.
        changeSupport.firePropertyChange("emailAddress", previousEmail, newEmailAddress);
    }

    public void setUsername(String newUsername) {
        String previousUsername = this.username;
        this.username = newUsername;
        changeSupport.firePropertyChange("username", previousUsername, newUsername);
    }

    public String getUsername() { return username; }
    public String getEmailAddress() { return emailAddress; }
}

// --- Observer 1: Sends a verification email when the email address changes ---
class EmailVerificationService implements PropertyChangeListener {
    @Override
    public void propertyChange(PropertyChangeEvent event) {
        // Only react to the property we care about — ignore everything else.
        if ("emailAddress".equals(event.getPropertyName())) {
            System.out.printf("[Email Service] Verification sent to new address: %s (was: %s)%n",
                event.getNewValue(), event.getOldValue());
        }
    }
}

// --- Observer 2: Audit log that tracks ALL profile changes ---
class ProfileAuditLog implements PropertyChangeListener {
    @Override
    public void propertyChange(PropertyChangeEvent event) {
        // This one listens to every property — the event tells it which one changed.
        System.out.printf("[Audit] Field '%s' changed: '%s' → '%s'%n",
            event.getPropertyName(), event.getOldValue(), event.getNewValue());
    }
}

public class UserProfileObserver {
    public static void main(String[] args) {
        UserProfile profile = new UserProfile("jsmith", "john@oldmail.com");

        EmailVerificationService emailService = new EmailVerificationService();
        ProfileAuditLog auditLog = new ProfileAuditLog();

        profile.addPropertyChangeListener(emailService);
        profile.addPropertyChangeListener(auditLog);

        System.out.println("--- Changing email address ---");
        profile.setEmailAddress("john@newmail.com");

        System.out.println("--- Changing username ---");
        profile.setUsername("johnsmith");

        System.out.println("--- Setting same email again (no event fired) ---");
        // PropertyChangeSupport is smart: it won't fire if old == new value.
        profile.setEmailAddress("john@newmail.com");
    }
}

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

// A thread-safe Subject using CopyOnWriteArrayList.
// Safe for concurrent registrations and notifications.
class SensorDataFeed {

    // CopyOnWriteArrayList: reads are lock-free, writes copy the internal array.
    // Perfect for observer lists where reads (notifications) dominate writes (registrations).
    private final List<SensorObserver> observers = new CopyOnWriteArrayList<>();
    private volatile double latestReading; // volatile ensures visibility across threads.

    public void register(SensorObserver observer) {
        observers.add(observer);
    }

    public void unregister(SensorObserver observer) {
        observers.remove(observer); // Safe to call from any thread.
    }

    public void publishReading(double sensorValue) {
        this.latestReading = sensorValue;
        // Even if another thread adds/removes an observer mid-iteration,
        // CopyOnWriteArrayList won't throw ConcurrentModificationException.
        for (SensorObserver observer : observers) {
            observer.onNewReading(sensorValue);
        }
    }
}

interface SensorObserver {
    void onNewReading(double value);
}

// A short-lived observer that properly unregisters itself when done.
class TemperatureMonitor implements SensorObserver {
    private final SensorDataFeed feed;
    private final String monitorId;

    public TemperatureMonitor(String monitorId, SensorDataFeed feed) {
        this.monitorId = monitorId;
        this.feed = feed;
        feed.register(this); // Register on construction.
    }

    @Override
    public void onNewReading(double value) {
        System.out.printf("[%s] Temperature reading: %.1f°C%n", monitorId, value);
    }

    // Always provide a cleanup method — this prevents the memory leak.
    public void shutdown() {
        feed.unregister(this);
        System.out.printf("[%s] Unregistered from feed.%n", monitorId);
    }
}

public class ThreadSafeEventBus {
    public static void main(String[] args) throws InterruptedException {
        SensorDataFeed temperatureSensor = new SensorDataFeed();

        TemperatureMonitor kitchenMonitor = new TemperatureMonitor("KitchenSensor", temperatureSensor);
        TemperatureMonitor serverRoomMonitor = new TemperatureMonitor("ServerRoom", temperatureSensor);

        // Simulate concurrent sensor readings from a background thread.
        ExecutorService sensorThread = Executors.newSingleThreadExecutor();
        sensorThread.submit(() -> {
            temperatureSensor.publishReading(22.5);
            temperatureSensor.publishReading(23.1);
        });

        sensorThread.shutdown();
        sensorThread.awaitTermination(2, TimeUnit.SECONDS);

        // Simulate the kitchen monitor going offline — must unregister to avoid memory leak.
        kitchenMonitor.shutdown();

        System.out.println("--- Reading after kitchen monitor unregistered ---");
        temperatureSensor.publishReading(24.0); // Only serverRoomMonitor receives this.
    }
}

// Push model interface — subject sends full data
interface ObserverPush {
    void update(String ticker, double price, long timestamp);
}

// Pull model interface — subject sends minimal notification
interface ObserverPull {
    void update(StockMarket subject);
}

// Concrete pull observer
class PriceDisplay implements ObserverPull {
    @Override
    public void update(StockMarket subject) {
        // Pull only what we need
        System.out.println("Current price: " + subject.getCurrentPrice());
    }
}

// Hybrid: push a small event object
class PriceChangeEvent {
    final String ticker;
    final double newPrice;
    final StockMarket source;
    // constructor...
}

interface ObserverHybrid {
    void onPriceChanged(PriceChangeEvent event);
}

import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

// --- Custom Event ---
// Extend ApplicationEvent (or use generic PayloadApplicationEvent)
public class OrderPlacedEvent extends ApplicationEvent {
    private final String orderId;
    private final double total;

    public OrderPlacedEvent(Object source, String orderId, double total) {
        super(source);
        this.orderId = orderId;
        this.total = total;
    }

    public String getOrderId() { return orderId; }
    public double getTotal() { return total; }
}

// --- Listener using @EventListener (modern way) ---
@Component
class OrderConfirmationService {

    @EventListener
    public void onOrderPlaced(OrderPlacedEvent event) {
        System.out.printf("[Confirmation] Sending email for order %s, total $%.2f%n",
            event.getOrderId(), event.getTotal());
    }
}

// --- Listener using ApplicationListener (traditional way) ---
@Component
class OrderAuditService implements ApplicationListener<OrderPlacedEvent> {

    @Override
    public void onApplicationEvent(OrderPlacedEvent event) {
        System.out.printf("[Audit] Order %s recorded in audit log%n", event.getOrderId());
    }
}

// --- Publisher ---
@Component
class OrderService {
    private final ApplicationEventPublisher publisher;

    public OrderService(ApplicationEventPublisher publisher) {
        this.publisher = publisher;
    }

    public void placeOrder(String orderId, double total) {
        // Business logic...
        publisher.publishEvent(new OrderPlacedEvent(this, orderId, total));
    }
}

// --- Async listener (optional) ---
@Component
class ShippingService {

    @EventListener
    @Async  // Requires @EnableAsync
    public void onOrderPlaced(OrderPlacedEvent event) {
        System.out.printf("[Shipping] Preparing shipment for order %s (async)%n", event.getOrderId());
    }
}

import org.junit.jupiter.api.Test;
import static org.mockito.Mockito.*;

class ObserverTest {

    @Test
    void observerIsNotifiedOnStateChange() {
        // Arrange
        StockMarket subject = new StockMarket("AAPL", 170.0);
        StockObserver mockObserver = mock(StockObserver.class);
        subject.addObserver(mockObserver);

        // Act
        subject.updatePrice(180.0);

        // Assert
        verify(mockObserver, times(1)).onPriceChanged("AAPL", 180.0);
    }

    @Test
    void observerIsNotNotifiedAfterRemoval() {
        StockMarket subject = new StockMarket("AAPL", 170.0);
        StockObserver mockObserver = mock(StockObserver.class);
        subject.addObserver(mockObserver);
        subject.removeObserver(mockObserver);

        subject.updatePrice(180.0);

        verify(mockObserver, never()).onPriceChanged(any(), anyDouble());
    }

    @Test
    void multipleObserversAllReceiveUpdates() {
        StockMarket subject = new StockMarket("AAPL", 170.0);
        StockObserver obs1 = mock(StockObserver.class);
        StockObserver obs2 = mock(StockObserver.class);
        subject.addObserver(obs1);
        subject.addObserver(obs2);

        subject.updatePrice(180.0);

        verify(obs1).onPriceChanged("AAPL", 180.0);
        verify(obs2).onPriceChanged("AAPL", 180.0);
    }

    @Test
    void noMemoryLeakAfterRemovingAllObservers() throws Exception {
        StockMarket subject = new StockMarket("AAPL", 170.0);
        StockObserver observer = new StockObserver() {
            @Override
            public void onPriceChanged(String s, double v) {}
        };
        subject.addObserver(observer);
        subject.removeObserver(observer);

        // Weak reference check (conceptual).
        // In practice, verify subject.observers is empty.
        // If using CopyOnWriteArrayList, we can check size.
        // For real leak detection, use a WeakReference test after GC.
        // This is a simplified version.
        System.gc();
        // If observer is not referenced elsewhere, it should be collectable.
        // But without WeakReference, we just verify list is empty.
        // That's enough for this test.
    }
}