User Hit count

AssertJ

public class AssertJ {

    private AssertJ(){
        throw new AssertionError("Invalid Access");
    }
    /**
     *
     * @param object which will satisfy following code
     * <pre>
     *               new Predicate&lt;T&gt;(){
     *                  public boolean test(T t){
     *                      return t!=null;
     *                  }
     *               }
     * </pre>
     * @param message Error message in case Predicate fails
     * @param <T> describe the type of the object
     * @throws IllegalArgumentException in case passed object is null
     */
    public  static <T> void notNull(T object, String message){
        assertTrue(t -> t!=null,object,message);
    }

    /**
     *
     * @param predicate {@link Predicate} to test the condition
     * @param object Object which will satisfy the predicate
     * @param message Error message
     * @param args Argument to format the message
     * @param <T> describe the type of the object
     * @throws IllegalArgumentException throws {@link IllegalArgumentException} in case Asserttion fails
     */
    public  static <T> void assertTrue(Predicate<? super T> predicate, T object, String message, Object... args){
        if(!predicate.test(object)){
            message = getFinalErrorMessage(message, args);
            if(message==null){
                throw new IllegalArgumentException();
            }else{
                throw  new IllegalArgumentException(message);
            }
        }
    }

    /**
     *
     * @param message Error Message in can be null or empty
     * @param args Arguments for format the error message
     * @return Empty Error Message in case this is null or empty or actual error message
     */
    private static String getFinalErrorMessage(String message, Object... args){
        String actualMessage=null;
        if(message!=null && !("").equals(message.trim())){
            message = message.trim();
            if(args!=null && args.length>0) {
                actualMessage = String.format(message, args);
            }else{
                actualMessage = message;
            }
        }
        return actualMessage;
    }
}

TimeWindow

import java.time.Instant;
import java.time.temporal.ChronoUnit;


/**
 * TimeWindow which will define difference between two {@link Instant}
 * @see TimeWindow#SECONDS
 * @see TimeWindow#MINUTES
 * @see TimeWindow#HOURS
 * @see TimeWindow#DAYS
 */
public enum TimeWindow {
    //Return time difference in seconds
    SECONDS(1000){
        @Override
        public long calculate(Instant t1, Instant t2) {
            checkValidity(t1, t2);
            return Math.round((double)ChronoUnit.MILLIS.between(t1, t2)/getUnitConversionFromMills());
        }
    },

    //Return time difference in minutes
    MINUTES(1000*60){
        @Override
        public long calculate(Instant t1, Instant t2) {
            checkValidity(t1, t2);
            return Math.round((double)ChronoUnit.MILLIS.between(t1, t2)/getUnitConversionFromMills());
        }
    },

    //Return time difference in hours
    HOURS(1000*60*60){
        @Override
        public long calculate(Instant t1, Instant t2) {
            checkValidity(t1, t2);
            return Math.round((double)ChronoUnit.MILLIS.between(t1, t2)/getUnitConversionFromMills());
        }
    },

    //Return time difference in days
    DAYS(1000*60*60*24){
        @Override
        public long calculate(Instant t1, Instant t2) {
            checkValidity(t1, t2);
            return Math.round((double)ChronoUnit.MILLIS.between(t1, t2)/getUnitConversionFromMills());
        }
    };

    private static void checkValidity(Instant t1, Instant t2){
        AssertJ.notNull(t1, "Duration can not be null");
        AssertJ.notNull(t2, "Duration can not be null");
    }

    private long unitConversionFromMills;

    TimeWindow(long unitConversionFromMills) {
        this.unitConversionFromMills = unitConversionFromMills;
    }

    public long getUnitConversionFromMills() {
        return unitConversionFromMills;
    }

    public abstract long calculate(final Instant t1, final Instant t2);


    public static void main(String[] args) throws InterruptedException {
        Instant i1 = Instant.now();
        Thread.sleep(2998L);
        Instant i2 = Instant.now();

        System.out.println(TimeWindow.SECONDS.calculate(i1,i2));
    }

}

User

public class User {
    private int userId;
    private Instant currentTime;

    public User(int userId) {
        this.userId = userId;
        currentTime = Instant.now();
    }

    public int getUserId() {
        return userId;
    }

    public Instant getCurrentTime() {
        return currentTime;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        User user = (User) o;
        return userId == user.userId;
    }

    @Override
    public int hashCode() {
        return Objects.hash(userId);
    }

    @Override
    public String toString() {
        return "User{" +
                "userId=" + userId +
                '}';
    }
}

UserHitCount

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

import java.util.concurrent.atomic.LongAdder;


public class UserHitCount {
    private ConcurrentMap<User, ConcurrentMap<Long, LongAdder>> userHitCount;
    private final long windowSize;
    private final TimeWindow unit;

    public UserHitCount(long windowSize, TimeWindow unit) {
        assert (windowSize>0);
        assert (unit!=null);
        userHitCount = new ConcurrentHashMap<>();
        this.windowSize = windowSize;
        this.unit = unit;
    }


    public  void visit(User u){
        System.out.println("Visited User "+u.getUserId());
        final Instant userStartTime = u.getCurrentTime();
        ConcurrentMap<Long,LongAdder> userHitsPerWindow = userHitCount.computeIfAbsent(u, user -> new ConcurrentHashMap<>());
        long difference = unit.calculate(userStartTime,Instant.now());
        if(!userHitsPerWindow.isEmpty()){
            long endUnitTobeRemoved = difference-windowSize;
            long startKey = userHitsPerWindow.keySet().iterator().next();
            for(long i=startKey;i<=endUnitTobeRemoved;i++){
                userHitsPerWindow.remove(i);
            }
        }
        LongAdder adder = userHitsPerWindow.computeIfAbsent(difference,d->new LongAdder());
        adder.add(1L);
        userHitsPerWindow.put(difference,adder);
        userHitCount.put(u,userHitsPerWindow);
    }

    public  long getCount(User u){
        ConcurrentMap<Long,LongAdder> userHitsPerWindow = userHitCount.get(u);
        if(userHitsPerWindow == null){
            return 0;
        }
        LongAdder adder = new LongAdder();
        userHitsPerWindow.values().parallelStream().forEach(
                v -> adder.add(v.sum())
        );
        return adder.sum();
    }

    public static void main(String[] args) throws InterruptedException {
        List<User> list = new ArrayList<>();
        UserHitCount count = new UserHitCount(500,TimeWindow.SECONDS);
        Random r = new Random();
        for(int i=0;i<10;i++){
            list.add(new User(i+1));
        }
        CyclicBarrier barrier = new CyclicBarrier(10, () -> {
            for(User u:list){
                System.out.println(String.format("user %s is visited %d times", u, count.getCount(u)));
            }
        });

        for(int i=0;i<10;i++){
            Thread t = new Thread(() -> {
                try {
                    User u = list.get(r.nextInt(list.size()-1));
                    count.visit(u);
                    barrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    throw new RuntimeException(e);
                }
            });
            t.start();
        }




    }
}