package io.realm.mongodb.sync; import io.realm.internal.Util; import io.realm.internal.util.Pair; import io.realm.log.RealmLog; import io.realm.mongodb.AppException; import io.realm.mongodb.ErrorCode; import io.realm.mongodb.User; import java.net.URI; import java.util.HashMap; import java.util.IdentityHashMap; import java.util.Iterator; import java.util.Locale; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; public class SyncSession { static final byte CONNECTION_VALUE_CONNECTED = 2; static final byte CONNECTION_VALUE_CONNECTING = 1; static final byte CONNECTION_VALUE_DISCONNECTED = 0; private static final int DIRECTION_DOWNLOAD = 1; private static final int DIRECTION_UPLOAD = 2; private static final byte STATE_VALUE_ACTIVE = 0; private static final byte STATE_VALUE_DYING = 1; private static final byte STATE_VALUE_INACTIVE = 2; private final long appNativePointer; private final ClientResetHandler clientResetHandler; private final SyncConfiguration configuration; private final CopyOnWriteArrayList<ConnectionListener> connectionListeners = new CopyOnWriteArrayList<>(); private final ErrorHandler errorHandler; private volatile boolean isClosed = false; private final Map<Long, Pair<ProgressListener, Progress>> listenerIdToProgressListenerMap = new HashMap(); private long nativeConnectionListenerToken; private final AtomicLong progressListenerId = new AtomicLong(-1); private final Map<ProgressListener, Long> progressListenerToOsTokenMap = new IdentityHashMap(); private final AtomicInteger waitCounter = new AtomicInteger(0); private final Object waitForChangesMutex = new Object(); private final AtomicReference<WaitForSessionWrapper> waitingForServerChanges = new AtomicReference<>(null); public interface ClientResetHandler { void onClientReset(SyncSession syncSession, ClientResetRequiredError clientResetRequiredError); } public interface ErrorHandler { void onError(SyncSession syncSession, AppException appException); } private native long nativeAddConnectionListener(long j, String str); private native long nativeAddProgressListener(long j, String str, long j2, int i, boolean z); private static native byte nativeGetConnectionState(long j, String str); private static native byte nativeGetState(long j, String str); private static native void nativeRemoveConnectionListener(long j, long j2, String str); private static native void nativeRemoveProgressListener(long j, String str, long j2); private static native void nativeShutdownAndWait(long j, String str); private static native void nativeStart(long j, String str); private static native void nativeStop(long j, String str); private native boolean nativeWaitForDownloadCompletion(long j, int i, String str); private native boolean nativeWaitForUploadCompletion(long j, int i, String str); public enum State { INACTIVE((byte) 2), ACTIVE((byte) 0), DYING((byte) 1); final byte value; private State(byte b) { this.value = b; } static State fromNativeValue(long j) { State[] values = values(); for (State state : values) { if (((long) state.value) == j) { return state; } } throw new IllegalArgumentException("Unknown session state code: " + j); } } SyncSession(SyncConfiguration syncConfiguration, long j) { this.configuration = syncConfiguration; this.errorHandler = syncConfiguration.getErrorHandler(); this.clientResetHandler = syncConfiguration.getClientResetHandler(); this.appNativePointer = j; } public SyncConfiguration getConfiguration() { return this.configuration; } public User getUser() { return this.configuration.getUser(); } public URI getServerUrl() { return this.configuration.getServerUrl(); } /* access modifiers changed from: package-private */ public void notifySessionError(String str, int i, String str2) { AppException appException; if (this.errorHandler != null) { ErrorCode fromNativeError = ErrorCode.fromNativeError(str, i); if (fromNativeError == ErrorCode.CLIENT_RESET) { this.clientResetHandler.onClientReset(this, new ClientResetRequiredError(this.appNativePointer, fromNativeError, "A Client Reset is required. Read more here: https://docs.realm.io/sync/using-synced-realms/errors#client-reset.", this.configuration, this.configuration.forErrorRecovery(str2))); return; } if (fromNativeError == ErrorCode.UNKNOWN) { appException = new AppException(str, i, str2); } else { appException = new AppException(fromNativeError, str2); } this.errorHandler.onError(this, appException); } } public State getState() { byte nativeGetState = nativeGetState(this.appNativePointer, this.configuration.getPath()); if (nativeGetState != -1) { return State.fromNativeValue((long) nativeGetState); } throw new IllegalStateException("Could not find session, Realm was probably closed"); } public ConnectionState getConnectionState() { byte nativeGetConnectionState = nativeGetConnectionState(this.appNativePointer, this.configuration.getPath()); if (nativeGetConnectionState != -1) { return ConnectionState.fromNativeValue((long) nativeGetConnectionState); } throw new IllegalStateException("Could not find session, Realm was probably closed"); } public boolean isConnected() { ConnectionState fromNativeValue = ConnectionState.fromNativeValue((long) nativeGetConnectionState(this.appNativePointer, this.configuration.getPath())); State state = getState(); return (state == State.ACTIVE || state == State.DYING) && fromNativeValue == ConnectionState.CONNECTED; } /* access modifiers changed from: package-private */ public synchronized void notifyProgressListener(long j, long j2, long j3) { Pair<ProgressListener, Progress> pair = this.listenerIdToProgressListenerMap.get(Long.valueOf(j)); if (pair != null) { S s = (S) new Progress(j2, j3); if (!s.equals(pair.second)) { pair.second = s; try { pair.first.onChange(s); } catch (Exception e) { RealmLog.error(e); } } } else { RealmLog.debug("Trying unknown listener failed: " + j, new Object[0]); } } /* access modifiers changed from: package-private */ public void notifyConnectionListeners(long j, long j2) { Iterator<ConnectionListener> it = this.connectionListeners.iterator(); while (it.hasNext()) { try { it.next().onChange(ConnectionState.fromNativeValue(j), ConnectionState.fromNativeValue(j2)); } catch (Exception e) { RealmLog.error(e); } } } public synchronized void addDownloadProgressListener(ProgressMode progressMode, ProgressListener progressListener) { addProgressListener(progressMode, 1, progressListener); } public synchronized void addUploadProgressListener(ProgressMode progressMode, ProgressListener progressListener) { addProgressListener(progressMode, 2, progressListener); } public synchronized void removeProgressListener(ProgressListener progressListener) { if (progressListener != null) { Long remove = this.progressListenerToOsTokenMap.remove(progressListener); if (remove != null) { Iterator<Map.Entry<Long, Pair<ProgressListener, Progress>>> it = this.listenerIdToProgressListenerMap.entrySet().iterator(); while (true) { if (it.hasNext()) { if (it.next().getValue().first.equals(progressListener)) { it.remove(); break; } } else { break; } } nativeRemoveProgressListener(this.appNativePointer, this.configuration.getPath(), remove.longValue()); } } } private void addProgressListener(ProgressMode progressMode, int i, ProgressListener progressListener) { checkProgressListenerArguments(progressMode, progressListener); boolean z = progressMode == ProgressMode.INDEFINITELY; long incrementAndGet = this.progressListenerId.incrementAndGet(); this.listenerIdToProgressListenerMap.put(Long.valueOf(incrementAndGet), new Pair<>(progressListener, null)); long nativeAddProgressListener = nativeAddProgressListener(this.appNativePointer, this.configuration.getPath(), incrementAndGet, i, z); if (nativeAddProgressListener == 0) { this.listenerIdToProgressListenerMap.remove(Long.valueOf(incrementAndGet)); } else { this.progressListenerToOsTokenMap.put(progressListener, Long.valueOf(nativeAddProgressListener)); } } private void checkProgressListenerArguments(ProgressMode progressMode, ProgressListener progressListener) { Util.checkNull(progressListener, "listener"); Util.checkNull(progressMode, "mode"); } public synchronized void addConnectionChangeListener(ConnectionListener connectionListener) { Util.checkNull(connectionListener, "listener"); if (this.connectionListeners.isEmpty()) { this.nativeConnectionListenerToken = nativeAddConnectionListener(this.appNativePointer, this.configuration.getPath()); } this.connectionListeners.add(connectionListener); } public synchronized void removeConnectionChangeListener(ConnectionListener connectionListener) { Util.checkNull(connectionListener, "listener"); this.connectionListeners.remove(connectionListener); if (this.connectionListeners.isEmpty()) { nativeRemoveConnectionListener(this.appNativePointer, this.nativeConnectionListenerToken, this.configuration.getPath()); } } /* access modifiers changed from: package-private */ public synchronized void close() { this.isClosed = true; } private void notifyAllChangesSent(int i, String str, Long l, String str2) { WaitForSessionWrapper waitForSessionWrapper = this.waitingForServerChanges.get(); if (waitForSessionWrapper != null && this.waitCounter.get() == i) { waitForSessionWrapper.handleResult(str, l, str2); } } public void downloadAllServerChanges() throws InterruptedException { Util.checkNotOnMainThread("downloadAllServerChanges() cannot be called from the main thread."); synchronized (this.waitForChangesMutex) { waitForChanges(1, Long.MAX_VALUE, TimeUnit.MILLISECONDS); } } public boolean downloadAllServerChanges(long j, TimeUnit timeUnit) throws InterruptedException { boolean waitForChanges; Util.checkNotOnMainThread("downloadAllServerChanges() cannot be called from the main thread."); checkTimeout(j, timeUnit); synchronized (this.waitForChangesMutex) { waitForChanges = waitForChanges(1, j, timeUnit); } return waitForChanges; } public void uploadAllLocalChanges() throws InterruptedException { Util.checkNotOnMainThread("uploadAllLocalChanges() cannot be called from the main thread."); synchronized (this.waitForChangesMutex) { waitForChanges(2, Long.MAX_VALUE, TimeUnit.MILLISECONDS); } } public boolean uploadAllLocalChanges(long j, TimeUnit timeUnit) throws InterruptedException { boolean waitForChanges; Util.checkNotOnMainThread("uploadAllLocalChanges() cannot be called from the main thread."); checkTimeout(j, timeUnit); synchronized (this.waitForChangesMutex) { waitForChanges = waitForChanges(2, j, timeUnit); } return waitForChanges; } public synchronized void start() { nativeStart(this.appNativePointer, this.configuration.getPath()); } public synchronized void stop() { close(); nativeStop(this.appNativePointer, this.configuration.getPath()); } private boolean waitForChanges(int i, long j, TimeUnit timeUnit) throws InterruptedException { boolean z; String str; if (i == 1 || i == 2) { boolean z2 = false; if (!this.isClosed) { String path = this.configuration.getPath(); WaitForSessionWrapper waitForSessionWrapper = new WaitForSessionWrapper(); this.waitingForServerChanges.set(waitForSessionWrapper); int incrementAndGet = this.waitCounter.incrementAndGet(); if (i == 1) { z = nativeWaitForDownloadCompletion(this.appNativePointer, incrementAndGet, path); } else { z = nativeWaitForUploadCompletion(this.appNativePointer, incrementAndGet, path); } if (!z) { this.waitingForServerChanges.set(null); if (i == 1) { str = "It was not possible to download all remote changes."; } else if (i != 2) { throw new IllegalArgumentException("Unknown direction: " + i); } else { str = "It was not possible upload all local changes."; } ErrorCode errorCode = ErrorCode.UNKNOWN; throw new AppException(errorCode, str + " Has the SyncClient been started?"); } try { z2 = waitForSessionWrapper.waitForServerChanges(j, timeUnit); try { if (!this.isClosed && !waitForSessionWrapper.isSuccess()) { waitForSessionWrapper.throwExceptionIfNeeded(); } } finally { this.waitingForServerChanges.set(null); } } catch (InterruptedException e) { this.waitingForServerChanges.set(null); throw e; } } return z2; } throw new IllegalArgumentException("Unknown direction: " + i); } private void checkTimeout(long j, TimeUnit timeUnit) { if (j <= 0) { throw new IllegalArgumentException("'timeout' must be > 0. It was: " + j); } else if (timeUnit == null) { throw new IllegalArgumentException("Non-null 'unit' required"); } } /* access modifiers changed from: package-private */ public void shutdownAndWait() { nativeShutdownAndWait(this.appNativePointer, this.configuration.getPath()); } /* access modifiers changed from: private */ public static class WaitForSessionWrapper { private String errorCategory; private Long errorCode; private String errorMessage; private volatile boolean resultReceived; private final CountDownLatch waiter; private WaitForSessionWrapper() { this.waiter = new CountDownLatch(1); this.resultReceived = false; this.errorCode = null; } public boolean waitForServerChanges(long j, TimeUnit timeUnit) throws InterruptedException { if (!this.resultReceived) { return this.waiter.await(j, timeUnit); } return isSuccess(); } public void handleResult(String str, Long l, String str2) { this.errorCategory = str; this.errorCode = l; this.errorMessage = str2; this.resultReceived = true; this.waiter.countDown(); } public boolean isSuccess() { return this.resultReceived && this.errorCode == null; } public void throwExceptionIfNeeded() { Long l; if (this.resultReceived && (l = this.errorCode) != null) { long longValue = l.longValue(); ErrorCode fromNativeError = ErrorCode.fromNativeError(this.errorCategory, (int) longValue); if (longValue < -2147483648L || longValue > 2147483647L || fromNativeError == ErrorCode.UNKNOWN) { throw new AppException(fromNativeError, String.format(Locale.US, "Internal error (%d): %s", this.errorCode, this.errorMessage)); } throw new AppException(fromNativeError, this.errorMessage); } } } }