0
\$\begingroup\$

(See the next iteration:
Delayed, concurrent event stack in Java - follow-up ) Motivation

I was confronted with a task of having "message events" for a GUI program. The use case is as follows: we have two events; A and B. A is visible for, say, 10 seconds; and B is visible for 5 seconds. Also, suppose that B triggers after 2 seconds after the A triggers. All in all, we have these steps:

  • A triggers at time 0 seconds,
  • B triggers at time 2 seconds,
  • B finishes at time 7 seconds,
  • A finishes at time 10 seconds.

com.github.coderodde.eventstack.DelayedEventStack.java:

package com.github.coderodde.eventstack;

import java.util.Deque;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * This class implements an delayed event stack.
 * 
 * @author Rodion "rodde" Efremov
 * @version 1.6 (Jan 12, 2023)
 * @since 1.6 (Jan 12, 2023)
 */
public final class DelayedEventStack implements AutoCloseable {

    /**
     * This static inner class implements the event stack entry.
     */
    public static final class DelayedEventStackEntry {
        private final Runnable onExpired;
        private final long expirationMillis;

        public DelayedEventStackEntry(Runnable onExpired,
                                      long entryMillis,
                                      long durationMillis) {
            this.onExpired = 
                    Objects.requireNonNull(onExpired, "onExpired is null.");

            this.expirationMillis = entryMillis + durationMillis;
        }
    }

    /**
     * The atomic boolean flag for running the event loop. 
     */
    private final AtomicBoolean doRunFlag = new AtomicBoolean(true);

    /**
     * The actual event stack.
     */
    private final Deque<DelayedEventStackEntry> delayedEventStack = 
            new ConcurrentLinkedDeque<>();

    /**
     * This flag specifies whether the stack runs the leftover events in the 
     * stack upon closing the stack.
     */
    private final boolean runExpiredOnClose;

    /** 
     * The worker thread.
     */
    private final Thread workerThread = new Thread() {
        @Override
        public void run() {
            while (doRunFlag.get()) {
                // Ask delayedEventStack for a topmost event entry. Gets null if 
                // the stack is empty, in which case we sleep a millisecond and 
                // ask one more time.
                DelayedEventStackEntry topmostEventStackEntry = 
                        delayedEventStack.peekLast();

                if (topmostEventStackEntry == null) {
                    // Once here, the stack is empty.
                    DelayedEventStack.sleep(1L);
                } else if (System.currentTimeMillis() >= 
                        topmostEventStackEntry.expirationMillis) {
                    // Once here, we can discharge an event from the stack.
                    topmostEventStackEntry.onExpired.run();
                    delayedEventStack.removeLast();
                } else {
                    // Don't abuse the CPU.
                    DelayedEventStack.sleep(1L);
                }
            }
        }
    };

    public DelayedEventStack() {
        this(true);
    }

    public DelayedEventStack(boolean runExpiredOnClose) {
        this.runExpiredOnClose = runExpiredOnClose;
        workerThread.start();
    }

    public void add(Runnable onAdd, Runnable onExpired, long durationMillis) {
        delayedEventStack.addLast(new DelayedEventStackEntry(
                                  onExpired, 
                                  System.currentTimeMillis(), 
                                  durationMillis));
        onAdd.run();
    }

    @Override
    public void close() throws Exception {
        while (!delayedEventStack.isEmpty()) {
            delayedEventStack.removeLast().onExpired.run();
        }

        doRunFlag.set(false);
    }

    public static void main(String[] args) throws Throwable {
        System.out.println("Simulation started.");

        DelayedEventStack eventStack = new DelayedEventStack();

        eventStack.add(() -> { System.out.println("Event 1 start"); },
                       () -> { System.out.println("Event 1 end");},
                       10_000L);

        sleep(2_000L);

        eventStack.add(() -> { System.out.println("Event 2 start"); },
                       () -> { System.out.println("Event 2 end");},
                       3000L);

        sleep(7_000L);

        eventStack.add(() -> { System.out.println("Leftover event start"); },
                       () -> { System.out.println("Leftover event end"); },
                       10_000L);

        CloseThread closeThread = new CloseThread(eventStack);
        closeThread.start();
    }

    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException ex) {}
    }

    private static final class CloseThread extends Thread {

        private final DelayedEventStack eventStack;

        CloseThread(DelayedEventStack eventStack) {
            this.eventStack = eventStack;
        }

        @Override
        public void run() {
            DelayedEventStack.sleep(10_000L);

            try {
                eventStack.close();
                System.out.println("Event stack closed.");
            } catch (Exception ex) {
                System.err.println(ex.getMessage());
            }
        }
    }
}

Output

Simulation started.
Event 1 start
Event 2 start
Event 2 end
Leftover event start
Leftover event end
Event 1 end
Event stack closed.

Critique request

As I am not proficient in concurrent computing, I need your help to make my event stack mature.

\$\endgroup\$

3 Answers 3

1
\$\begingroup\$

doRunFlag is an instance variable volatile is redundant.

The behaviour of DelayedEventStack#add method suggests the DelayedEventStack might be modelled through java.util.Observable.

CloseThread's running delay fits for being an initialisation parameter.

What java.util.concurrent.ScheduledThreadPoolExecutor implements seems similar with the implementation of DelayedEventStack.

\$\endgroup\$
1
  • 1
    \$\begingroup\$ The code doesn't contain volatile anywhere, so I really can't make sense of your first paragraph. What did you mean to say? \$\endgroup\$ Commented Jan 17, 2023 at 8:21
0
\$\begingroup\$

From the OOP perspective boolean arguments are subject for improvements. A way to follow object oriented programming principles - abstraction, inheritance, encapsulation and polymorphism - could be to implement design patterns; for the actual use case Factory and Command design patterns.

Factory by providing methods creating differently initialised DelayedEventStack instances.

Command by implementing optional behaviour decoupled from DelayedEventStack's class behaviour.

Code snippet relevant for early mentioned details, although there are alternatives for event driven architecture included in Java language:

import java.util.Deque;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;

public final class DelayedEventStack implements AutoCloseable {

    /**
     * This static inner class implements the event stack entry.
     */
    public static final class DelayedEventStackEntry {
        public final Runnable onExpired;
        private final long expirationMillis;

        public DelayedEventStackEntry(Runnable onExpired, long entryMillis, long durationMillis) {
            this.onExpired = Objects.requireNonNull(onExpired, "onExpired is null.");

            this.expirationMillis = entryMillis + durationMillis;
        }
    }

    /**
     * Support class for optional behaviour.
     */
    private class StackEvent {

        private final Set<DequeCommand> commands;

        StackEvent() {
            this.commands = new HashSet<>();
        }

        void on() {
            for (DequeCommand command : this.commands) {
                command.on();
            }
        }

        boolean add(DequeCommand toAdd) {
            return this.commands.add(toAdd);
        }
    }

    private interface DequeCommand {

        void on();
    }

     private static class DelayedStackEventCloseCommand implements DequeCommand {

        final Deque<DelayedEventStackEntry> deque;

        DelayedStackEventCloseCommand(Deque<DelayedEventStackEntry> deque) {
            this.deque = deque;
        }

        @Override
        public void on() {
            while (!this.deque.isEmpty()) {
                this.deque.removeLast().onExpired.run();
            }
        }
    }

     /** 
      * Factory method 
      */
    public static DelayedEventStack instanceSelfCleaning() {

        final DelayedEventStack eventStack = DelayedEventStack.instance();
        eventStack.closeEvent.add(new DelayedStackEventCloseCommand(eventStack.delayedEventStack));

        return eventStack;
    }

    /** 
     * Factory method 
     */
    public static DelayedEventStack instance() {
        return new DelayedEventStack();
    }

    /**
     * The actual event stack.
     */
    public final Deque<DelayedEventStackEntry> delayedEventStack;

    final StackEvent closeEvent;

    /**
     * The worker thread.
     */
    private final Thread workerThread = new Thread() {
        @Override
        public void run() {
            .
            .
            .
        }
    };

    private DelayedEventStack() {
        this.delayedEventStack = new ConcurrentLinkedDeque<>();
        this.closeEvent = new StackEvent();
    }

    public void add(Runnable onAdd, Runnable onExpired, long durationMillis) {
        delayedEventStack.addLast(new DelayedEventStackEntry(onExpired, System.currentTimeMillis(), durationMillis));
        onAdd.run();
    }

    @Override
    public void close() throws Exception {
        workerThread.interrupt();
        this.closeEvent.on();
    }

    public static void main(String[] args) throws Throwable {
        System.out.println("Simulation started.");

        DelayedEventStack eventStack =  DelayedEventStack.instanceSelfCleaning();
        .
        .
        .
    }

    static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException ex) { } }

    private static final class CloseThread extends Thread {
        .
        .
        .
    }
}
\$\endgroup\$
0
\$\begingroup\$

Using concurrency package with the purpose of interrupting a non-concurrent running application is over head.

The way boolean variables are used in the implementation are flags. In object oriented programming polymorphism is preferred over flag programming.

The Java language includes java.lang.Object#wait methods and java.util.Timer class are convenient for the implemented scenario.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.