Specialized Data Structures

/**
 * Get a distributed atomic long for thread-safe long operations
 * @param name - unique name of the atomic long
 * @return RAtomicLong instance
 */
public RAtomicLong getAtomicLong(String name);
public RAtomicLong getAtomicLong(CommonOptions options);

/**
 * Get a distributed atomic double for thread-safe double operations  
 * @param name - unique name of the atomic double
 * @return RAtomicDouble instance
 */
public RAtomicDouble getAtomicDouble(String name);
public RAtomicDouble getAtomicDouble(CommonOptions options);

/**
 * Get a long adder for high-performance distributed counting
 * @param name - unique name of the long adder
 * @return RLongAdder instance optimized for concurrent additions
 */
public RLongAdder getLongAdder(String name);
public RLongAdder getLongAdder(CommonOptions options);

/**
 * Get a double adder for high-performance distributed summing
 * @param name - unique name of the double adder  
 * @return RDoubleAdder instance optimized for concurrent additions
 */
public RDoubleAdder getDoubleAdder(String name);
public RDoubleAdder getDoubleAdder(CommonOptions options);

// Distributed atomic long with compare-and-swap operations
public interface RAtomicLong extends RObject, RExpirable, RAtomicLongAsync {
    long get();
    void set(long newValue);
    long getAndSet(long newValue);
    
    // Compare and swap operations
    boolean compareAndSet(long expect, long update);
    boolean weakCompareAndSet(long expect, long update);
    
    // Arithmetic operations
    long addAndGet(long delta);
    long getAndAdd(long delta);
    long incrementAndGet();
    long getAndIncrement();
    long decrementAndGet();
    long getAndDecrement();
    
    // Functional operations
    long updateAndGet(LongUnaryOperator updateFunction);
    long getAndUpdate(LongUnaryOperator updateFunction);
    long accumulateAndGet(long x, LongBinaryOperator accumulatorFunction);
    long getAndAccumulate(long x, LongBinaryOperator accumulatorFunction);
}

// Distributed atomic double
public interface RAtomicDouble extends RObject, RExpirable, RAtomicDoubleAsync {
    double get();
    void set(double newValue);
    double getAndSet(double newValue);
    
    boolean compareAndSet(double expect, double update);
    double addAndGet(double delta);
    double getAndAdd(double delta);
    
    // Functional operations
    double updateAndGet(DoubleUnaryOperator updateFunction);
    double getAndUpdate(DoubleUnaryOperator updateFunction);
    double accumulateAndGet(double x, DoubleBinaryOperator accumulatorFunction);
    double getAndAccumulate(double x, DoubleBinaryOperator accumulatorFunction);
}

// High-performance distributed counter
public interface RLongAdder extends RObject, RExpirable, RLongAdderAsync {
    void add(long x);
    void increment();
    void decrement();
    long sum();
    void reset();
    long sumThenReset();
    String toString();
}

// High-performance distributed double accumulator
public interface RDoubleAdder extends RObject, RExpirable, RDoubleAdderAsync {
    void add(double x);
    double sum();
    void reset();
    double sumThenReset();
    String toString();
}

// Atomic long operations
RAtomicLong counter = redisson.getAtomicLong("globalCounter");
counter.set(100);

// Thread-safe operations across multiple JVMs
long newValue = counter.incrementAndGet(); // 101
long previous = counter.getAndAdd(10);     // returns 101, sets to 111

// Compare and swap for conditional updates
long current = counter.get();
boolean success = counter.compareAndSet(current, current * 2);
if (success) {
    System.out.println("Successfully doubled the counter");
}

// Functional updates
counter.updateAndGet(value -> value > 200 ? 0 : value + 1);
counter.accumulateAndGet(5, (current, delta) -> Math.max(current, delta));

// High-performance counting with RLongAdder
RLongAdder adder = redisson.getLongAdder("requestCounter");

// Multiple threads can increment concurrently with high performance
adder.increment();
adder.add(5);

// Get sum when needed (more expensive operation)
long totalRequests = adder.sum();
System.out.println("Total requests: " + totalRequests);

// Reset counter
long finalCount = adder.sumThenReset(); // Get value and reset atomically

// Atomic double for precise calculations
RAtomicDouble balance = redisson.getAtomicDouble("accountBalance");
balance.set(1000.50);

double newBalance = balance.addAndGet(-50.25); // Withdraw $50.25
boolean withdrawSuccess = balance.compareAndSet(950.25, 900.25); // Conditional withdraw

/**
 * Get a Bloom filter for membership testing
 * @param name - unique name of the Bloom filter
 * @return RBloomFilter instance for approximate membership testing
 */
public <V> RBloomFilter<V> getBloomFilter(String name);
public <V> RBloomFilter<V> getBloomFilter(String name, Codec codec);
public <V> RBloomFilter<V> getBloomFilter(PlainOptions options);

/**
 * Get a HyperLogLog for cardinality estimation
 * @param name - unique name of the HyperLogLog
 * @return RHyperLogLog instance for approximate unique count
 */
public <V> RHyperLogLog<V> getHyperLogLog(String name);
public <V> RHyperLogLog<V> getHyperLogLog(String name, Codec codec);
public <V> RHyperLogLog<V> getHyperLogLog(PlainOptions options);

// Bloom filter for membership testing
public interface RBloomFilter<T> extends RObject, RExpirable, RBloomFilterAsync<T> {
    // Filter initialization
    boolean tryInit(long expectedInsertions, double falseProbability);
    boolean isExists();
    
    // Membership operations
    boolean contains(T object);
    boolean add(T object);
    
    // Bulk operations
    boolean[] contains(T... objects);
    boolean addAll(Collection<T> c);
    boolean containsAll(Collection<T> c);
    
    // Filter statistics
    long count();
    long getExpectedInsertions();
    double getFalseProbability();
    long getHashIterations();
    long getSize();
}

// HyperLogLog for cardinality estimation
public interface RHyperLogLog<V> extends RObject, RExpirable, RHyperLogLogAsync<V> {
    // Add elements for counting unique values
    boolean add(V obj);
    boolean addAll(Collection<V> objects);
    
    // Get approximate unique count
    long count();
    
    // Merge with other HyperLogLogs
    long countWith(String... otherLogNames);
    void mergeWith(String... otherLogNames);
}

// Bloom filter for membership testing
RBloomFilter<String> bloomFilter = redisson.getBloomFilter("uniqueUsers");

// Initialize filter with expected elements and false positive rate
bloomFilter.tryInit(10000, 0.03); // 10K elements, 3% false positive rate

// Add elements
bloomFilter.add("user123");
bloomFilter.add("user456");
bloomFilter.add("user789");

// Test membership
boolean containsUser = bloomFilter.contains("user123"); // true
boolean containsNew = bloomFilter.contains("user999");  // false (or small chance of false positive)

// Bulk operations
Collection<String> users = Arrays.asList("user1", "user2", "user3");
bloomFilter.addAll(users);
boolean allExist = bloomFilter.containsAll(users); // true

// Check filter statistics
long elementCount = bloomFilter.count();
double falsePositiveRate = bloomFilter.getFalseProbability();
System.out.println("Filter contains ~" + elementCount + " elements with " + 
                   (falsePositiveRate * 100) + "% false positive rate");

// HyperLogLog for unique count estimation
RHyperLogLog<String> uniqueVisitors = redisson.getHyperLogLog("dailyVisitors");

// Add visitor IDs (can be millions with constant memory usage)
uniqueVisitors.add("visitor1");
uniqueVisitors.add("visitor2");
uniqueVisitors.add("visitor1"); // Duplicate - won't affect count

// Add many visitors
for (int i = 0; i < 100000; i++) {
    uniqueVisitors.add("visitor" + i);
}

// Get approximate unique count (very memory efficient)
long uniqueCount = uniqueVisitors.count();
System.out.println("Approximate unique visitors: " + uniqueCount);

// Merge multiple HyperLogLogs
RHyperLogLog<String> weeklyVisitors = redisson.getHyperLogLog("weeklyVisitors");
RHyperLogLog<String> monthlyVisitors = redisson.getHyperLogLog("monthlyVisitors");

// Count across multiple logs without merging
long totalUniqueVisitors = uniqueVisitors.countWith("weeklyVisitors", "monthlyVisitors");

// Merge logs into one
uniqueVisitors.mergeWith("weeklyVisitors", "monthlyVisitors");

/**
 * Get a geo-spatial collection for location operations
 * @param name - unique name of the geo collection
 * @return RGeo instance for geospatial operations
 */
public <V> RGeo<V> getGeo(String name);
public <V> RGeo<V> getGeo(String name, Codec codec);
public <V> RGeo<V> getGeo(PlainOptions options);

// Geospatial operations interface
public interface RGeo<V> extends RObject, RExpirable, RGeoAsync<V> {
    // Add locations
    long add(double longitude, double latitude, V member);
    long add(GeoEntry... entries);
    
    // Distance calculations
    Double dist(V firstMember, V secondMember, GeoUnit geoUnit);
    
    // Get coordinates
    Map<V, GeoPosition> pos(V... members);
    List<String> hash(V... members);
    
    // Radius searches
    List<V> radius(double longitude, double latitude, double radius, GeoUnit geoUnit);
    List<V> radius(double longitude, double latitude, double radius, GeoUnit geoUnit, int count);
    List<V> radius(double longitude, double latitude, double radius, GeoUnit geoUnit, 
                   GeoOrder geoOrder);
    List<V> radius(double longitude, double latitude, double radius, GeoUnit geoUnit, 
                   GeoOrder geoOrder, int count);
    
    // Radius searches with distance
    Map<V, Double> radiusWithDistance(double longitude, double latitude, double radius, 
                                     GeoUnit geoUnit);
    Map<V, Double> radiusWithDistance(double longitude, double latitude, double radius, 
                                     GeoUnit geoUnit, int count);
    Map<V, Double> radiusWithDistance(double longitude, double latitude, double radius, 
                                     GeoUnit geoUnit, GeoOrder geoOrder);
    
    // Radius searches with position  
    Map<V, GeoPosition> radiusWithPosition(double longitude, double latitude, double radius, 
                                          GeoUnit geoUnit);
    
    // Member-based radius searches
    List<V> radius(V member, double radius, GeoUnit geoUnit);
    Map<V, Double> radiusWithDistance(V member, double radius, GeoUnit geoUnit);
    Map<V, GeoPosition> radiusWithPosition(V member, double radius, GeoUnit geoUnit);
    
    // Search and store results
    long radiusStoreTo(String destName, double longitude, double latitude, double radius, 
                       GeoUnit geoUnit);
    long radiusStoreTo(String destName, V member, double radius, GeoUnit geoUnit);
}

// Geospatial data types
public class GeoEntry {
    private final double longitude;
    private final double latitude;
    private final Object member;
    
    public GeoEntry(double longitude, double latitude, Object member);
    // getters...
}

public class GeoPosition {
    private final double longitude;
    private final double latitude;
    
    public GeoPosition(double longitude, double latitude);
    // getters...
}

public enum GeoUnit {
    METERS("m"),
    KILOMETERS("km"), 
    MILES("mi"),
    FEET("ft");
    
    private final String value;
    GeoUnit(String value) { this.value = value; }
}

public enum GeoOrder {
    ASC, DESC
}

// Geospatial operations for location services
RGeo<String> locations = redisson.getGeo("cityLocations");

// Add locations (longitude, latitude, member)
locations.add(-74.0059, 40.7128, "New York");      // NYC
locations.add(-118.2437, 34.0522, "Los Angeles");  // LA
locations.add(-87.6298, 41.8781, "Chicago");       // Chicago
locations.add(-122.4194, 37.7749, "San Francisco"); // SF

// Calculate distance between cities
Double distance = locations.dist("New York", "Los Angeles", GeoUnit.MILES);
System.out.println("Distance NY to LA: " + distance + " miles");

// Get coordinates of a city
Map<String, GeoPosition> positions = locations.pos("New York", "Chicago");
GeoPosition nycPos = positions.get("New York");
System.out.println("NYC: " + nycPos.getLongitude() + ", " + nycPos.getLatitude());

// Find cities within radius of a point (around NYC)
List<String> nearbyCities = locations.radius(-74.0059, 40.7128, 300, GeoUnit.MILES);
System.out.println("Cities within 300 miles of NYC: " + nearbyCities);

// Find cities with distances
Map<String, Double> citiesWithDistance = locations.radiusWithDistance(-74.0059, 40.7128, 
                                                                      500, GeoUnit.MILES);
citiesWithDistance.forEach((city, dist) -> {
    System.out.println(city + " is " + dist + " miles away");
});

// Find cities near another city
List<String> nearChicago = locations.radius("Chicago", 200, GeoUnit.MILES);
System.out.println("Cities within 200 miles of Chicago: " + nearChicago);

// Store search results in another geo collection
long stored = locations.radiusStoreTo("eastCoastCities", -74.0059, 40.7128, 
                                     300, GeoUnit.MILES);
System.out.println("Stored " + stored + " east coast cities");

// Delivery service example
RGeo<String> deliveryPoints = redisson.getGeo("deliveryLocations");

// Add delivery addresses
deliveryPoints.add(-73.935242, 40.730610, "customer1"); // East Village
deliveryPoints.add(-73.989308, 40.756800, "customer2"); // Times Square
deliveryPoints.add(-73.958805, 40.768923, "customer3"); // Upper East Side

// Find deliveries within 2 miles of delivery truck location
List<String> nearbyDeliveries = deliveryPoints.radius(-73.950000, 40.750000, 
                                                      2, GeoUnit.MILES);
System.out.println("Nearby deliveries: " + nearbyDeliveries);

// Restaurant finder example
RGeo<Restaurant> restaurants = redisson.getGeo("restaurants");
restaurants.add(-73.9857, 40.7484, new Restaurant("Pizza Palace", "Italian"));
restaurants.add(-73.9759, 40.7505, new Restaurant("Burger Barn", "American"));
restaurants.add(-73.9851, 40.7589, new Restaurant("Sushi Spot", "Japanese"));

// Find restaurants near user location, sorted by distance
Map<Restaurant, Double> nearbyRestaurants = restaurants.radiusWithDistance(
    -73.9800, 40.7500, 0.5, GeoUnit.MILES, GeoOrder.ASC);
    
nearbyRestaurants.forEach((restaurant, distance) -> {
    System.out.println(restaurant.getName() + " - " + distance + " miles");
});

/**
 * Get a distributed bit set for binary operations
 * @param name - unique name of the bit set
 * @return RBitSet instance for bit manipulation
 */
public RBitSet getBitSet(String name);
public RBitSet getBitSet(CommonOptions options);

/**
 * Get a binary stream for raw binary data operations
 * @param name - unique name of the binary stream
 * @return RBinaryStream instance for binary I/O
 */
public RBinaryStream getBinaryStream(String name);
public RBinaryStream getBinaryStream(CommonOptions options);

// Distributed bit set for binary operations
public interface RBitSet extends RObject, RExpirable, RBitSetAsync {
    // Bit operations
    boolean get(long bitIndex);
    boolean getAndSet(long bitIndex);
    void set(long bitIndex);
    void set(long bitIndex, boolean value);
    void set(long fromIndex, long toIndex);
    void set(long fromIndex, long toIndex, boolean value);
    
    // Bit clearing
    void clear();
    void clear(long bitIndex);
    void clear(long fromIndex, long toIndex);
    
    // Bit testing
    boolean isEmpty();
    long length();
    long size();
    long cardinality(); // Count of set bits
    
    // Bit operations with other bit sets
    void and(String... bitSetNames);
    void or(String... bitSetNames);
    void xor(String... bitSetNames);
    void not();
    
    // Bit searching
    long nextSetBit(long fromIndex);
    long nextClearBit(long fromIndex);
    long previousSetBit(long fromIndex);
    long previousClearBit(long fromIndex);
    
    // Bulk operations
    byte[] toByteArray();
    BitSet asBitSet();
}

// Binary stream for raw data
public interface RBinaryStream extends RObject, RBinaryStreamAsync {
    // Stream operations
    InputStream getInputStream();
    OutputStream getOutputStream();
    
    // Direct read/write
    void set(byte[] value);
    void set(InputStream inputStream);
    byte[] get();
    
    // Partial operations
    byte[] get(long offset, long length);
    void set(byte[] value, long offset);
    
    // Stream information
    long size();
    boolean delete();
}

// Bit set operations for flags and permissions
RBitSet permissions = redisson.getBitSet("userPermissions");

// Set permission bits (bit index represents permission ID)
permissions.set(0);  // Read permission
permissions.set(1);  // Write permission
permissions.set(5);  // Admin permission

// Check permissions
boolean canRead = permissions.get(0);   // true
boolean canWrite = permissions.get(1);  // true
boolean canDelete = permissions.get(2); // false

// Set multiple bits
permissions.set(10, 15); // Set bits 10-14

// Count set permissions
long activePermissions = permissions.cardinality();
System.out.println("User has " + activePermissions + " permissions");

// Combine permissions from multiple users
RBitSet user1Perms = redisson.getBitSet("user1Permissions");
RBitSet user2Perms = redisson.getBitSet("user2Permissions");

user1Perms.set(0, 5); // Basic permissions
user2Perms.set(3, 8); // Different permissions

// Create combined permission set
RBitSet combinedPerms = redisson.getBitSet("combinedPermissions");
combinedPerms.or("user1Permissions", "user2Permissions");

// Find next available permission slot
long nextAvailable = combinedPerms.nextClearBit(0);
System.out.println("Next available permission ID: " + nextAvailable);

// Binary stream for file-like operations
RBinaryStream fileData = redisson.getBinaryStream("uploadedFile");

// Store binary data
byte[] imageData = loadImageFromDisk();
fileData.set(imageData);

// Read binary data
byte[] retrievedData = fileData.get();
saveImageToDisk(retrievedData);

// Stream operations for large files
try (InputStream input = new FileInputStream("largefile.zip");
     OutputStream output = fileData.getOutputStream()) {
    
    byte[] buffer = new byte[8192];
    int bytesRead;
    while ((bytesRead = input.read(buffer)) != -1) {
        output.write(buffer, 0, bytesRead);
    }
}

// Read part of binary data
long fileSize = fileData.size();
byte[] header = fileData.get(0, 1024); // First 1KB
byte[] footer = fileData.get(fileSize - 1024, 1024); // Last 1KB

// Update part of binary data
byte[] newHeader = createNewHeader();
fileData.set(newHeader, 0); // Replace header

System.out.println("File size: " + fileData.size() + " bytes");

/**
 * Get an ID generator for distributed unique ID creation
 * @param name - unique name of the ID generator
 * @return RIdGenerator instance
 */
public RIdGenerator getIdGenerator(String name);
public RIdGenerator getIdGenerator(CommonOptions options);

public interface RIdGenerator extends RObject, RExpirable, RIdGeneratorAsync {
    // Initialize generator with allocation size
    boolean tryInit(long value, long allocationSize);
    
    // Generate next ID
    long nextId();
    
    // Current state
    long current();
    
    // Allocation settings
    long getAllocationSize();
    boolean expire(long timeToLive, TimeUnit timeUnit);
}

// Distributed ID generation
RIdGenerator orderIds = redisson.getIdGenerator("orderIdGenerator");

// Initialize with starting value and allocation size
orderIds.tryInit(10000, 100); // Start at 10000, allocate 100 IDs at a time

// Generate unique order IDs
long orderId1 = orderIds.nextId(); // 10000
long orderId2 = orderIds.nextId(); // 10001
long orderId3 = orderIds.nextId(); // 10002

System.out.println("Generated order IDs: " + orderId1 + ", " + orderId2 + ", " + orderId3);

// Multiple ID generators for different entity types
RIdGenerator userIds = redisson.getIdGenerator("userIdGenerator");
RIdGenerator productIds = redisson.getIdGenerator("productIdGenerator");

userIds.tryInit(1000, 50);
productIds.tryInit(5000, 25);

// Each generator maintains independent sequences
long userId = userIds.nextId();     // 1000
long productId = productIds.nextId(); // 5000

// Check current values
long currentUser = userIds.current();
long currentProduct = productIds.current();

System.out.println("Current user ID: " + currentUser);
System.out.println("Current product ID: " + currentProduct);

// High-performance ID generation in multi-threaded environment
ExecutorService executor = Executors.newFixedThreadPool(10);
RIdGenerator threadSafeIds = redisson.getIdGenerator("threadSafeGenerator");
threadSafeIds.tryInit(0, 1000); // Large allocation for high throughput

// Multiple threads generating IDs concurrently
for (int i = 0; i < 10; i++) {
    executor.submit(() -> {
        for (int j = 0; j < 1000; j++) {
            long id = threadSafeIds.nextId();
            System.out.println("Thread " + Thread.currentThread().getName() + 
                             " generated ID: " + id);
        }
    });
}

executor.shutdown();

/**
 * Get a rate limiter for distributed rate limiting
 * @param name - unique name of the rate limiter
 * @return RRateLimiter instance
 */
public RRateLimiter getRateLimiter(String name);
public RRateLimiter getRateLimiter(CommonOptions options);

public interface RRateLimiter extends RObject, RExpirable, RRateLimiterAsync {
    // Initialize rate limiter
    boolean trySetRate(RateType type, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);
    
    // Acquire permits
    boolean tryAcquire();
    boolean tryAcquire(long permits);
    boolean tryAcquire(long timeout, TimeUnit unit);
    boolean tryAcquire(long permits, long timeout, TimeUnit unit);
    
    void acquire();
    void acquire(long permits);
    
    // Rate limiter information
    long availablePermits();
    RateLimiterConfig getConfig();
}

public enum RateType {
    OVERALL,      // Total rate across all instances
    PER_CLIENT    // Rate per client instance
}

public enum RateIntervalUnit {
    SECONDS, MINUTES, HOURS, DAYS
}

// API rate limiting
RRateLimiter apiLimiter = redisson.getRateLimiter("apiRequests");

// Allow 100 requests per minute across all instances
apiLimiter.trySetRate(RateType.OVERALL, 100, 1, RateIntervalUnit.MINUTES);

// Check if request is allowed
if (apiLimiter.tryAcquire()) {
    // Process API request
    processApiRequest();
} else {
    // Rate limit exceeded
    throw new RateLimitExceededException("Too many requests");
}

// Per-user rate limiting
RRateLimiter userLimiter = redisson.getRateLimiter("user:" + userId + ":requests");
userLimiter.trySetRate(RateType.PER_CLIENT, 10, 1, RateIntervalUnit.MINUTES);

// Acquire multiple permits for batch operations
if (userLimiter.tryAcquire(5, 1, TimeUnit.SECONDS)) {
    processBatchRequest(5);
}

// Blocking acquire with automatic rate limiting
userLimiter.acquire(); // Blocks until permit is available
processRequest();

/**
 * Get a time series for time-ordered data storage
 * @param name - unique name of the time series
 * @return RTimeSeries instance
 */
public <V, L> RTimeSeries<V, L> getTimeSeries(String name);
public <V, L> RTimeSeries<V, L> getTimeSeries(String name, Codec codec);
public <V, L> RTimeSeries<V, L> getTimeSeries(PlainOptions options);

public interface RTimeSeries<V, L> extends RObject, RExpirable, RTimeSeriesAsync<V, L> {
    // Add data points
    void add(long timestamp, V object);
    void add(long timestamp, V object, L label);
    void addAll(Map<Long, V> objects);
    
    // Range queries
    Collection<TimeSeriesEntry<V, L>> range(long startTimestamp, long endTimestamp);
    Collection<TimeSeriesEntry<V, L>> range(long startTimestamp, long endTimestamp, int count);
    Collection<TimeSeriesEntry<V, L>> rangeReversed(long startTimestamp, long endTimestamp);
    
    // Get entries by count
    Collection<TimeSeriesEntry<V, L>> first(int count);
    Collection<TimeSeriesEntry<V, L>> last(int count);
    
    // Size and info
    long size();
    long size(long startTimestamp, long endTimestamp);
    TimeSeriesEntry<V, L> firstEntry();
    TimeSeriesEntry<V, L> lastEntry();
    
    // Remove operations
    boolean remove(long timestamp);
    long removeRange(long startTimestamp, long endTimestamp);
    
    // Polling operations
    Collection<TimeSeriesEntry<V, L>> pollFirst(int count);
    Collection<TimeSeriesEntry<V, L>> pollLast(int count);
}

public class TimeSeriesEntry<V, L> {
    private final long timestamp;
    private final V value;
    private final L label;
    
    // constructors and getters...
}

// System metrics time series
RTimeSeries<Double, String> cpuMetrics = redisson.getTimeSeries("cpu_usage");

// Add CPU usage data points
long now = System.currentTimeMillis();
cpuMetrics.add(now, 45.5, "server1");
cpuMetrics.add(now + 60000, 52.3, "server1");  // 1 minute later
cpuMetrics.add(now + 120000, 38.7, "server1"); // 2 minutes later

// Query metrics for the last hour
long oneHourAgo = now - (60 * 60 * 1000);
Collection<TimeSeriesEntry<Double, String>> recentMetrics = 
    cpuMetrics.range(oneHourAgo, now);

recentMetrics.forEach(entry -> {
    System.out.println("Time: " + entry.getTimestamp() + 
                      ", CPU: " + entry.getValue() + "%" +
                      ", Server: " + entry.getLabel());
});

// Stock price time series
RTimeSeries<BigDecimal, String> stockPrices = redisson.getTimeSeries("AAPL_prices");

// Add stock price data
stockPrices.add(now, new BigDecimal("150.25"), "NASDAQ");
stockPrices.add(now + 300000, new BigDecimal("151.10"), "NASDAQ"); // 5 minutes later

// Get latest 10 price points
Collection<TimeSeriesEntry<BigDecimal, String>> latestPrices = stockPrices.last(10);

// Remove old data (older than 30 days)
long thirtyDaysAgo = now - (30L * 24 * 60 * 60 * 1000);
long removedCount = stockPrices.removeRange(0, thirtyDaysAgo);
System.out.println("Removed " + removedCount + " old entries");

/**
 * Get a Redis stream for message streaming
 * @param name - unique name of the stream
 * @return RStream instance
 */
public <K, V> RStream<K, V> getStream(String name);
public <K, V> RStream<K, V> getStream(String name, Codec codec);
public <K, V> RStream<K, V> getStream(PlainOptions options);

public interface RStream<K, V> extends RObject, RExpirable, RStreamAsync<K, V> {
    // Add messages
    StreamMessageId add(StreamAddArgs<K, V> args);
    StreamMessageId add(K key, V value);
    StreamMessageId add(Map<K, V> entries);
    
    // Read messages
    Map<String, Map<StreamMessageId, Map<K, V>>> read(StreamReadArgs args);
    Map<StreamMessageId, Map<K, V>> read(StreamMessageId id);
    Map<StreamMessageId, Map<K, V>> read(int count, StreamMessageId id);
    
    // Range operations
    Map<StreamMessageId, Map<K, V>> range(StreamMessageId startId, StreamMessageId endId);
    Map<StreamMessageId, Map<K, V>> range(int count, StreamMessageId startId, StreamMessageId endId);
    Map<StreamMessageId, Map<K, V>> rangeReversed(StreamMessageId startId, StreamMessageId endId);
    
    // Consumer groups
    void createGroup(String groupName);
    void createGroup(String groupName, StreamMessageId id);
    
    // Stream info
    long size();
    StreamInfo<K, V> getInfo();
    
    // Remove operations
    long remove(StreamMessageId... ids);
    long trim(StreamTrimArgs args);
}

public class StreamMessageId {
    public static final StreamMessageId NEWEST = new StreamMessageId("$");
    public static final StreamMessageId ALL = new StreamMessageId("0");
    
    // constructors and methods...
}

// Event streaming
RStream<String, Object> eventStream = redisson.getStream("user_events");

// Add events to stream
Map<String, Object> userLogin = new HashMap<>();
userLogin.put("event", "login");
userLogin.put("userId", "user123");
userLogin.put("timestamp", System.currentTimeMillis());
userLogin.put("ip", "192.168.1.100");

StreamMessageId loginId = eventStream.add(userLogin);
System.out.println("Added login event with ID: " + loginId);

// Add purchase event
StreamMessageId purchaseId = eventStream.add("event", "purchase", 
                                           "userId", "user123",
                                           "productId", "prod456",
                                           "amount", 99.99);

// Read all messages from beginning
Map<StreamMessageId, Map<String, Object>> allEvents = 
    eventStream.read(StreamMessageId.ALL);

allEvents.forEach((id, event) -> {
    System.out.println("Event ID: " + id + ", Data: " + event);
});

// Create consumer group for processing
eventStream.createGroup("analytics_processor");

// Chat messaging with streams
RStream<String, String> chatStream = redisson.getStream("chat:room1");

// Send messages
chatStream.add("user", "alice", "message", "Hello everyone!");
chatStream.add("user", "bob", "message", "Hi Alice!");
chatStream.add("user", "charlie", "message", "Good morning!");

// Read latest messages
Map<StreamMessageId, Map<String, String>> recentMessages = 
    chatStream.read(10, StreamMessageId.NEWEST);

// Process messages
recentMessages.forEach((id, msg) -> {
    String user = msg.get("user");
    String message = msg.get("message");
    System.out.println("[" + user + "]: " + message);
});

// Log processing with streams
RStream<String, Object> logStream = redisson.getStream("application_logs");

// Add log entries
logStream.add("level", "ERROR", 
             "message", "Database connection failed",
             "component", "UserService",
             "timestamp", System.currentTimeMillis());

logStream.add("level", "INFO",
             "message", "User registered successfully", 
             "userId", "user789",
             "timestamp", System.currentTimeMillis());

// Create consumer group for log processing
logStream.createGroup("log_processor");

// Trim old logs (keep last 1000 entries)
long trimmed = logStream.trim(StreamTrimArgs.maxLen(1000));
System.out.println("Trimmed " + trimmed + " old log entries");

// Combining data structures for complex operations
public class AdvancedPatterns {
    
    // Rate limiting with atomic operations
    public static boolean isRequestAllowed(RedissonClient redisson, String clientId, 
                                         int maxRequests, int windowSeconds) {
        RAtomicLong requestCount = redisson.getAtomicLong("rate_limit:" + clientId);
        
        long current = requestCount.get();
        if (current == 0) {
            // First request, set expiration
            requestCount.set(1);
            requestCount.expire(windowSeconds, TimeUnit.SECONDS);
            return true;
        } else if (current < maxRequests) {
            requestCount.incrementAndGet();
            return true;
        } else {
            return false; // Rate limit exceeded
        }
    }
    
    // Caching with Bloom filter pre-check
    public static <T> T getWithBloomFilter(RedissonClient redisson, String key, 
                                          Class<T> type, Supplier<T> loader) {
        RBloomFilter<String> cacheFilter = redisson.getBloomFilter("cache_filter");
        
        // Quick check if key might exist
        if (!cacheFilter.contains(key)) {
            return null; // Definitely not in cache
        }
        
        // Might exist, check actual cache
        RBucket<T> bucket = redisson.getBucket(key);
        T value = bucket.get();
        
        if (value == null && loader != null) {
            // Load and cache
            value = loader.get();
            if (value != null) {
                bucket.set(value);
                cacheFilter.add(key);
            }
        }
        
        return value;
    }
    
    // Geospatial + time-series for location tracking
    public static void trackLocation(RedissonClient redisson, String userId, 
                                   double longitude, double latitude) {
        // Store in geospatial index
        RGeo<String> locations = redisson.getGeo("user_locations");
        locations.add(longitude, latitude, userId);
        
        // Store in time series for history
        RTimeSeries<GeoPosition, String> locationHistory = 
            redisson.getTimeSeries("location_history:" + userId);
        locationHistory.add(System.currentTimeMillis(), new GeoPosition(longitude, latitude));
        
        // Update atomic counter
        RAtomicLong locationUpdates = redisson.getAtomicLong("location_updates:" + userId);
        locationUpdates.incrementAndGet();
    }
}