package com.mongodb.internal.connection;
import com.mongodb.async.SingleResultCallback;
import com.mongodb.connection.ConnectionDescription;
import com.mongodb.diagnostics.logging.Logger;
import com.mongodb.diagnostics.logging.Loggers;
import com.mongodb.session.SessionContext;
import org.bson.ByteBuf;
import org.bson.codecs.Decoder;
import java.util.List;
import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
class UsageTrackingInternalConnection implements InternalConnection {
private static final Logger LOGGER = Loggers.getLogger("connection");
private volatile long openedAt;
private volatile long lastUsedAt;
private final int generation;
private final InternalConnection wrapped;
UsageTrackingInternalConnection(final InternalConnection wrapped, final int generation) {
this.wrapped = wrapped;
this.generation = generation;
openedAt = Long.MAX_VALUE;
lastUsedAt = openedAt;
}
@Override
public void open() {
wrapped.open();
openedAt = System.currentTimeMillis();
lastUsedAt = openedAt;
}
@Override
public void openAsync(final SingleResultCallback<Void> callback) {
wrapped.openAsync(new SingleResultCallback<Void>() {
@Override
public void onResult(final Void result, final Throwable t) {
if (t != null) {
callback.onResult(null, t);
} else {
openedAt = System.currentTimeMillis();
lastUsedAt = openedAt;
callback.onResult(null, null);
}
}
});
}
@Override
public void close() {
wrapped.close();
}
@Override
public boolean opened() {
return wrapped.opened();
}
@Override
public boolean isClosed() {
return wrapped.isClosed();
}
@Override
public ByteBuf getBuffer(final int size) {
return wrapped.getBuffer(size);
}
@Override
public void sendMessage(final List<ByteBuf> byteBuffers, final int lastRequestId) {
wrapped.sendMessage(byteBuffers, lastRequestId);
lastUsedAt = System.currentTimeMillis();
}
@Override
public <T> T sendAndReceive(final CommandMessage message, final Decoder<T> decoder, final SessionContext sessionContext) {
T result = wrapped.sendAndReceive(message, decoder, sessionContext);
lastUsedAt = System.currentTimeMillis();
return result;
}
@Override
public <T> void sendAndReceiveAsync(final CommandMessage message, final Decoder<T> decoder,
final SessionContext sessionContext, final SingleResultCallback<T> callback) {
SingleResultCallback<T> errHandlingCallback = errorHandlingCallback(new SingleResultCallback<T>() {
@Override
public void onResult(final T result, final Throwable t) {
lastUsedAt = System.currentTimeMillis();
callback.onResult(result, t);
}
}, LOGGER);
wrapped.sendAndReceiveAsync(message, decoder, sessionContext, errHandlingCallback);
}
@Override
public ResponseBuffers receiveMessage(final int responseTo) {
ResponseBuffers responseBuffers = wrapped.receiveMessage(responseTo);
lastUsedAt = System.currentTimeMillis();
return responseBuffers;
}
@Override
public void sendMessageAsync(final List<ByteBuf> byteBuffers, final int lastRequestId, final SingleResultCallback<Void> callback) {
SingleResultCallback<Void> errHandlingCallback = errorHandlingCallback(new SingleResultCallback<Void>() {
@Override
public void onResult(final Void result, final Throwable t) {
lastUsedAt = System.currentTimeMillis();
callback.onResult(result, t);
}
}, LOGGER);
wrapped.sendMessageAsync(byteBuffers, lastRequestId, errHandlingCallback);
}
@Override
public void receiveMessageAsync(final int responseTo, final SingleResultCallback<ResponseBuffers> callback) {
SingleResultCallback<ResponseBuffers> errHandlingCallback = errorHandlingCallback(new SingleResultCallback<ResponseBuffers>() {
@Override
public void onResult(final ResponseBuffers result, final Throwable t) {
lastUsedAt = System.currentTimeMillis();
callback.onResult(result, t);
}
}, LOGGER);
wrapped.receiveMessageAsync(responseTo, errHandlingCallback);
}
@Override
public ConnectionDescription getDescription() {
return wrapped.getDescription();
}
int getGeneration() {
return generation;
}
long getOpenedAt() {
return openedAt;
}
long getLastUsedAt() {
return lastUsedAt;
}
}