Subscription System

/**
 * Subscription request interface for setting up real-time data monitoring
 */
public interface PlcSubscriptionRequest extends PlcSubscriptionTagRequest {
    /**
     * Execute the subscription request asynchronously
     * @return CompletableFuture containing the subscription response
     */
    CompletableFuture<? extends PlcSubscriptionResponse> execute();
    
    /**
     * Builder interface for constructing subscription requests
     */
    interface Builder extends PlcRequestBuilder {
        /**
         * Set global consumer for all subscription events
         * @param consumer Consumer function to handle PlcSubscriptionEvent
         * @return Builder instance for method chaining
         */
        Builder setConsumer(Consumer<PlcSubscriptionEvent> consumer);
        
        /**
         * Add cyclic subscription by tag address
         * @param name Logical name for the subscription
         * @param tagAddress PLC-specific tag address string
         * @param duration Polling interval
         * @return Builder instance for method chaining
         */
        Builder addCyclicTagAddress(String name, String tagAddress, Duration duration);
        
        /**
         * Add cyclic subscription by tag address with specific consumer
         * @param name Logical name for the subscription
         * @param tagAddress PLC-specific tag address string
         * @param duration Polling interval
         * @param consumer Consumer function for this specific subscription
         * @return Builder instance for method chaining
         */
        Builder addCyclicTagAddress(String name, String tagAddress, Duration duration, Consumer<PlcSubscriptionEvent> consumer);
        
        /**
         * Add cyclic subscription with pre-parsed tag
         * @param name Logical name for the subscription
         * @param tag PlcSubscriptionTag instance
         * @param duration Polling interval
         * @return Builder instance for method chaining
         */
        Builder addCyclicTag(String name, PlcSubscriptionTag tag, Duration duration);
        
        /**
         * Add cyclic subscription with pre-parsed tag and specific consumer
         * @param name Logical name for the subscription
         * @param tag PlcSubscriptionTag instance
         * @param duration Polling interval
         * @param consumer Consumer function for this specific subscription
         * @return Builder instance for method chaining
         */
        Builder addCyclicTag(String name, PlcSubscriptionTag tag, Duration duration, Consumer<PlcSubscriptionEvent> consumer);
        
        /**
         * Add change-of-state subscription by tag address
         * @param name Logical name for the subscription
         * @param tagAddress PLC-specific tag address string
         * @return Builder instance for method chaining
         */
        Builder addChangeOfStateTagAddress(String name, String tagAddress);
        
        /**
         * Add change-of-state subscription by tag address with specific consumer
         * @param name Logical name for the subscription
         * @param tagAddress PLC-specific tag address string
         * @param consumer Consumer function for this specific subscription
         * @return Builder instance for method chaining
         */
        Builder addChangeOfStateTagAddress(String name, String tagAddress, Consumer<PlcSubscriptionEvent> consumer);
        
        /**
         * Add change-of-state subscription with pre-parsed tag
         * @param name Logical name for the subscription
         * @param tag PlcSubscriptionTag instance
         * @return Builder instance for method chaining
         */
        Builder addChangeOfStateTag(String name, PlcSubscriptionTag tag);
        
        /**
         * Add change-of-state subscription with pre-parsed tag and specific consumer
         * @param name Logical name for the subscription
         * @param tag PlcSubscriptionTag instance
         * @param consumer Consumer function for this specific subscription
         * @return Builder instance for method chaining
         */
        Builder addChangeOfStateTag(String name, PlcSubscriptionTag tag, Consumer<PlcSubscriptionEvent> consumer);
        
        /**
         * Add event subscription by tag address
         * @param name Logical name for the subscription
         * @param tagAddress PLC-specific tag address string
         * @return Builder instance for method chaining
         */
        Builder addEventTagAddress(String name, String tagAddress);
        
        /**
         * Add event subscription by tag address with specific consumer
         * @param name Logical name for the subscription
         * @param tagAddress PLC-specific tag address string
         * @param consumer Consumer function for this specific subscription
         * @return Builder instance for method chaining
         */
        Builder addEventTagAddress(String name, String tagAddress, Consumer<PlcSubscriptionEvent> consumer);
        
        /**
         * Add event subscription with pre-parsed tag
         * @param name Logical name for the subscription
         * @param tag PlcSubscriptionTag instance
         * @return Builder instance for method chaining
         */
        Builder addEventTag(String name, PlcSubscriptionTag tag);
        
        /**
         * Add event subscription with pre-parsed tag and specific consumer
         * @param name Logical name for the subscription
         * @param tag PlcSubscriptionTag instance
         * @param consumer Consumer function for this specific subscription
         * @return Builder instance for method chaining
         */
        Builder addEventTag(String name, PlcSubscriptionTag tag, Consumer<PlcSubscriptionEvent> consumer);
        
        /**
         * Build the subscription request
         * @return PlcSubscriptionRequest instance ready for execution
         */
        PlcSubscriptionRequest build();
    }
}

/**
 * Subscription response interface providing handles for managing active subscriptions
 */
public interface PlcSubscriptionResponse extends PlcSubscriptionTagResponse {
    /**
     * Get the originating subscription request
     * @return PlcSubscriptionRequest that generated this response
     */
    PlcSubscriptionRequest getRequest();
    
    /**
     * Get subscription handle by name for managing the subscription
     * @param name Subscription name from request
     * @return PlcSubscriptionHandle for managing the subscription
     */
    PlcSubscriptionHandle getSubscriptionHandle(String name);
    
    /**
     * Get all subscription handles
     * @return Collection of all PlcSubscriptionHandle instances
     */
    Collection<PlcSubscriptionHandle> getSubscriptionHandles();
}

/**
 * Subscription event interface extending PlcReadResponse with timestamp
 */
public interface PlcSubscriptionEvent extends PlcReadResponse {
    /**
     * Get the timestamp when the event occurred
     * @return Instant representing the event timestamp
     */
    Instant getTimestamp();
}

/**
 * Handle for managing active subscriptions
 */
public interface PlcSubscriptionHandle {
    /**
     * Register an additional consumer for this subscription
     * @param consumer Consumer function to handle subscription events
     * @return PlcConsumerRegistration for managing the consumer
     */
    PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer);
}

/**
 * Unsubscription request interface for canceling active subscriptions
 */
public interface PlcUnsubscriptionRequest extends PlcRequest {
    /**
     * Execute the unsubscription request asynchronously
     * @return CompletableFuture containing the unsubscription response
     */
    CompletableFuture<? extends PlcUnsubscriptionResponse> execute();
    
    /**
     * Builder interface for constructing unsubscription requests
     */
    interface Builder extends PlcRequestBuilder {
        /**
         * Add subscription handle to unsubscribe
         * @param subscriptionHandle Handle from PlcSubscriptionResponse
         * @return Builder instance for method chaining
         */
        Builder addHandles(PlcSubscriptionHandle... subscriptionHandle);
        
        /**
         * Build the unsubscription request
         * @return PlcUnsubscriptionRequest instance ready for execution
         */
        PlcUnsubscriptionRequest build();
    }
}

/**
 * Unsubscription response interface
 */
public interface PlcUnsubscriptionResponse extends PlcResponse {
    PlcUnsubscriptionRequest getRequest();
}

import org.apache.plc4x.java.DefaultPlcDriverManager;
import org.apache.plc4x.java.api.PlcConnection;
import org.apache.plc4x.java.api.messages.*;
import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
import org.apache.plc4x.java.api.types.PlcResponseCode;
import java.time.Duration;
import java.util.function.Consumer;

// Cyclic subscription with global consumer
PlcDriverManager driverManager = new DefaultPlcDriverManager();
try (PlcConnection connection = driverManager.getConnection("modbus-tcp://192.168.1.100:502")) {
    connection.connect();
    
    // Global event consumer
    Consumer<PlcSubscriptionEvent> globalConsumer = event -> {
        System.out.println("Event at " + event.getTimestamp());
        for (String tagName : event.getTagNames()) {
            if (event.getResponseCode(tagName) == PlcResponseCode.OK) {
                System.out.println(tagName + ": " + event.getInteger(tagName));
            }
        }
    };
    
    // Create cyclic subscription (poll every 1 second)
    PlcSubscriptionRequest subscriptionRequest = connection.subscriptionRequestBuilder()
        .setConsumer(globalConsumer)
        .addCyclicTagAddress("temperature", "holding-register:1", Duration.ofSeconds(1))
        .addCyclicTagAddress("pressure", "holding-register:2", Duration.ofSeconds(1))
        .build();
    
    PlcSubscriptionResponse response = subscriptionRequest.execute().get();
    
    // Keep subscription active
    Thread.sleep(30000); // Monitor for 30 seconds
    
    // Clean up subscriptions
    PlcUnsubscriptionRequest unsubRequest = connection.unsubscriptionRequestBuilder()
        .addHandles(response.getSubscriptionHandles().toArray(new PlcSubscriptionHandle[0]))
        .build();
    
    unsubRequest.execute().get();
}

// Individual consumer per subscription
try (PlcConnection connection = driverManager.getConnection("s7://192.168.1.200/0/1")) {
    connection.connect();
    
    // Individual consumers for different data types
    Consumer<PlcSubscriptionEvent> temperatureConsumer = event -> {
        if (event.getResponseCode("temperature") == PlcResponseCode.OK) {
            float temp = event.getFloat("temperature");
            System.out.println("Temperature: " + temp + "°C");
            
            // Temperature-specific logic
            if (temp > 80.0f) {
                System.out.println("HIGH TEMPERATURE ALERT!");
            }
        }
    };
    
    Consumer<PlcSubscriptionEvent> alarmConsumer = event -> {
        if (event.getResponseCode("alarm") == PlcResponseCode.OK) {
            boolean alarm = event.getBoolean("alarm");
            if (alarm) {
                System.err.println("ALARM TRIGGERED at " + event.getTimestamp());
            }
        }
    };
    
    PlcSubscriptionRequest subscriptionRequest = connection.subscriptionRequestBuilder()
        .addCyclicTagAddress("temperature", "DB1.DBD0:REAL", Duration.ofSeconds(2), temperatureConsumer)
        .addChangeOfStateTagAddress("alarm", "DB1.DBX10.0:BOOL", alarmConsumer)
        .build();
    
    PlcSubscriptionResponse response = subscriptionRequest.execute().get();
    
    // Monitor subscriptions
    Thread.sleep(60000); // Monitor for 1 minute
}

// Change-of-state subscription
try (PlcConnection connection = driverManager.getConnection("modbus-tcp://192.168.1.100:502")) {
    connection.connect();
    
    Consumer<PlcSubscriptionEvent> statusConsumer = event -> {
        System.out.println("Status changed at " + event.getTimestamp());
        for (String tagName : event.getTagNames()) {
            if (event.getResponseCode(tagName) == PlcResponseCode.OK) {
                boolean status = event.getBoolean(tagName);
                System.out.println(tagName + " is now " + (status ? "ON" : "OFF"));
            }
        }
    };
    
    PlcSubscriptionRequest subscriptionRequest = connection.subscriptionRequestBuilder()
        .addChangeOfStateTagAddress("motor1", "coil:1", statusConsumer)
        .addChangeOfStateTagAddress("motor2", "coil:2", statusConsumer)
        .addChangeOfStateTagAddress("valve1", "coil:10", statusConsumer)
        .build();
    
    PlcSubscriptionResponse response = subscriptionRequest.execute().get();
    
    // Keep monitoring
    Thread.sleep(120000); // Monitor for 2 minutes
}

// Event subscription for alarm conditions
try (PlcConnection connection = driverManager.getConnection("s7://192.168.1.200/0/1")) {
    connection.connect();
    
    Consumer<PlcSubscriptionEvent> eventConsumer = event -> {
        System.out.println("PLC Event occurred at " + event.getTimestamp());
        
        // Handle event data
        for (String tagName : event.getTagNames()) {
            PlcResponseCode code = event.getResponseCode(tagName);
            if (code == PlcResponseCode.OK) {
                // Process event based on tag name
                switch (tagName) {
                    case "emergency_stop":
                        boolean emergencyStop = event.getBoolean(tagName);
                        if (emergencyStop) {
                            System.err.println("EMERGENCY STOP ACTIVATED!");
                        }
                        break;
                    case "high_temperature":
                        boolean highTemp = event.getBoolean(tagName);
                        if (highTemp) {
                            System.err.println("HIGH TEMPERATURE EVENT!");
                        }
                        break;
                    case "low_pressure":
                        boolean lowPressure = event.getBoolean(tagName);
                        if (lowPressure) {
                            System.err.println("LOW PRESSURE EVENT!");
                        }
                        break;
                }
            }
        }
    };
    
    PlcSubscriptionRequest subscriptionRequest = connection.subscriptionRequestBuilder()
        .addEventTagAddress("emergency_stop", "DB100.DBX0.0:BOOL", eventConsumer)
        .addEventTagAddress("high_temperature", "DB100.DBX0.1:BOOL", eventConsumer)
        .addEventTagAddress("low_pressure", "DB100.DBX0.2:BOOL", eventConsumer)
        .build();
    
    PlcSubscriptionResponse response = subscriptionRequest.execute().get();
    
    // Keep event monitoring active
    Thread.sleep(300000); // Monitor for 5 minutes
}

// Dynamic subscription management
try (PlcConnection connection = driverManager.getConnection("modbus-tcp://192.168.1.100:502")) {
    connection.connect();
    
    Consumer<PlcSubscriptionEvent> dynamicConsumer = event -> {
        System.out.println("Dynamic event: " + event.getTimestamp());
        // Process event...
    };
    
    // Initial subscription
    PlcSubscriptionRequest initialRequest = connection.subscriptionRequestBuilder()
        .addCyclicTagAddress("sensor1", "holding-register:1", Duration.ofSeconds(1), dynamicConsumer)
        .build();
    
    PlcSubscriptionResponse initialResponse = initialRequest.execute().get();
    
    // Add more consumers to existing subscription
    PlcSubscriptionHandle sensor1Handle = initialResponse.getSubscriptionHandle("sensor1");
    PlcConsumerRegistration additionalConsumer = sensor1Handle.register(event -> {
        System.out.println("Additional processing for sensor1");
    });
    
    Thread.sleep(10000); // Monitor with additional consumer
    
    // Add new subscription dynamically
    PlcSubscriptionRequest additionalRequest = connection.subscriptionRequestBuilder()
        .addCyclicTagAddress("sensor2", "holding-register:2", Duration.ofSeconds(2), dynamicConsumer)
        .build();
    
    PlcSubscriptionResponse additionalResponse = additionalRequest.execute().get();
    
    Thread.sleep(20000); // Monitor both subscriptions
    
    // Clean up all subscriptions
    PlcUnsubscriptionRequest cleanup = connection.unsubscriptionRequestBuilder()
        .addHandles(initialResponse.getSubscriptionHandles().toArray(new PlcSubscriptionHandle[0]))
        .addHandles(additionalResponse.getSubscriptionHandles().toArray(new PlcSubscriptionHandle[0]))
        .build();
    
    cleanup.execute().get();
    System.out.println("All subscriptions cleaned up");
}

public enum PlcSubscriptionType {
    /**
     * Cyclic subscription - polls at regular intervals
     */
    CYCLIC,
    
    /**
     * Change-of-state subscription - triggers when value changes
     */
    CHANGE_OF_STATE,
    
    /**
     * Event subscription - triggers on PLC events
     */
    EVENT
}

public interface PlcSubscriptionTag extends PlcTag {
    // Extends PlcTag with subscription-specific behavior
}

public interface PlcConsumerRegistration {
    // Handle for managing consumer registration lifecycle
}

public interface PlcSubscriptionTagRequest extends PlcRequest {
    CompletableFuture<? extends PlcSubscriptionTagResponse> execute();
}

public interface PlcSubscriptionTagResponse extends PlcResponse {
    PlcSubscriptionTagRequest getRequest();
}