Object Store

/**
 * Store an object in the object store.
 * @param obj The Java object to be stored
 * @return ObjectRef instance representing the stored object
 */
public static <T> ObjectRef<T> put(T obj);

/**
 * Store an object with specified ownership.
 * @param obj The Java object to be stored
 * @param owner The actor that should own this object
 * @return ObjectRef instance representing the stored object
 */
public static <T> ObjectRef<T> put(T obj, BaseActorHandle owner);

import io.ray.api.Ray;
import io.ray.api.ObjectRef;

// Store simple objects
ObjectRef<String> stringRef = Ray.put("Hello Ray!");
ObjectRef<Integer> numberRef = Ray.put(42);

// Store complex objects
List<String> data = Arrays.asList("item1", "item2", "item3");
ObjectRef<List<String>> listRef = Ray.put(data);

// Store with actor ownership
ActorHandle<MyActor> actor = Ray.actor(MyActor::new).remote();
ObjectRef<String> ownedRef = Ray.put("owned-data", actor);

/**
 * Get an object by ObjectRef from the object store.
 * Blocks until the object is available.
 * @param objectRef The reference of the object to get
 * @return The Java object
 */
public static <T> T get(ObjectRef<T> objectRef);

/**
 * Get an object with timeout.
 * @param objectRef The reference of the object to get
 * @param timeoutMs Maximum time to wait in milliseconds
 * @return The Java object
 * @throws RayTimeoutException if timeout is exceeded
 */
public static <T> T get(ObjectRef<T> objectRef, long timeoutMs);

/**
 * Get multiple objects.
 * @param objectList List of object references
 * @return List of Java objects in the same order
 */
public static <T> List<T> get(List<ObjectRef<T>> objectList);

/**
 * Get multiple objects with timeout.
 * @param objectList List of object references
 * @param timeoutMs Maximum time to wait in milliseconds
 * @return List of Java objects in the same order
 * @throws RayTimeoutException if timeout is exceeded
 */
public static <T> List<T> get(List<ObjectRef<T>> objectList, long timeoutMs);

// Single object retrieval
ObjectRef<String> ref = Ray.put("Hello");
String value = Ray.get(ref);
System.out.println(value); // "Hello"

// Retrieval with timeout
try {
    String result = Ray.get(ref, 5000); // Wait max 5 seconds
} catch (RayTimeoutException e) {
    System.out.println("Object not ready within timeout");
}

// Multiple object retrieval
List<ObjectRef<Integer>> refs = Arrays.asList(
    Ray.put(1), Ray.put(2), Ray.put(3)
);
List<Integer> values = Ray.get(refs);
System.out.println(values); // [1, 2, 3]

/**
 * Wait for objects to be available.
 * @param waitList List of object references to wait for
 * @param numReturns Number of objects that should be ready
 * @param timeoutMs Maximum time to wait in milliseconds
 * @param fetchLocal If true, fetch objects locally before returning
 * @return WaitResult containing ready and unready object lists
 */
public static <T> WaitResult<T> wait(
    List<ObjectRef<T>> waitList, 
    int numReturns, 
    int timeoutMs, 
    boolean fetchLocal
);

/**
 * Wait for objects locally (fetchLocal = true).
 * @param waitList List of object references to wait for
 * @param numReturns Number of objects that should be ready
 * @param timeoutMs Maximum time to wait in milliseconds
 * @return WaitResult containing ready and unready object lists
 */
public static <T> WaitResult<T> wait(
    List<ObjectRef<T>> waitList, 
    int numReturns, 
    int timeoutMs
);

/**
 * Wait for objects with no timeout.
 * @param waitList List of object references to wait for
 * @param numReturns Number of objects that should be ready
 * @return WaitResult containing ready and unready object lists
 */
public static <T> WaitResult<T> wait(
    List<ObjectRef<T>> waitList, 
    int numReturns
);

/**
 * Wait for all objects.
 * @param waitList List of object references to wait for
 * @return WaitResult containing ready and unready object lists
 */
public static <T> WaitResult<T> wait(List<ObjectRef<T>> waitList);

// Start multiple long-running tasks
List<ObjectRef<String>> taskRefs = Arrays.asList(
    Ray.task(MyClass::longTask, "task1").remote(),
    Ray.task(MyClass::longTask, "task2").remote(),
    Ray.task(MyClass::longTask, "task3").remote()
);

// Wait for any 2 tasks to complete within 10 seconds
WaitResult<String> result = Ray.wait(taskRefs, 2, 10000);

System.out.println("Ready tasks: " + result.getReady().size());
System.out.println("Unready tasks: " + result.getUnready().size());

// Process completed tasks
for (ObjectRef<String> readyRef : result.getReady()) {
    String taskResult = Ray.get(readyRef);
    System.out.println("Task completed: " + taskResult);
}

// Wait for all remaining tasks
if (!result.getUnready().isEmpty()) {
    WaitResult<String> remaining = Ray.wait(result.getUnready());
    // Process remaining results...
}

public interface ObjectRef<T> {
    /**
     * Get the object (blocking).
     * @return The object value
     */
    T get();
    
    /**
     * Get the object with timeout.
     * @param timeoutMs Maximum time to wait in milliseconds
     * @return The object value
     * @throws RayTimeoutException if timeout is exceeded
     */
    T get(long timeoutMs);
}

ObjectRef<String> ref = Ray.put("data");

// Using ObjectRef interface directly
String value1 = ref.get();
String value2 = ref.get(5000);

// ObjectRefs can be passed around
public void processRef(ObjectRef<String> ref) {
    String data = ref.get();
    // Process data...
}

public class WaitResult<T> {
    /**
     * Get list of object references that are ready.
     * @return List of ready ObjectRef instances
     */
    public List<ObjectRef<T>> getReady();
    
    /**
     * Get list of object references that are not ready.
     * @return List of unready ObjectRef instances
     */
    public List<ObjectRef<T>> getUnready();
}

// Store large objects to avoid copying
List<LargeObject> bigData = createLargeDataset();
ObjectRef<List<LargeObject>> dataRef = Ray.put(bigData);

// Pass reference instead of data
ObjectRef<ProcessedData> result = Ray.task(MyProcessor::process, dataRef).remote();

try {
    ObjectRef<String> ref = Ray.put("data");
    String value = Ray.get(ref, 5000);
} catch (RayTimeoutException e) {
    System.out.println("Object not ready: " + e.getMessage());
} catch (RayException e) {
    System.out.println("Ray error: " + e.getMessage());
}

// Create actor
ActorHandle<DataProcessor> processor = Ray.actor(DataProcessor::new).remote();

// Store data with actor ownership
ObjectRef<String> data = Ray.put("important-data", processor);

// Data will be cleaned up when actor dies
processor.task(DataProcessor::shutdown).remote();