package org.apache.cassandra.tracing;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Stopwatch;
import org.slf4j.helpers.MessageFormatter;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventNotifier;
import org.apache.cassandra.utils.progress.ProgressListener;
public abstract class TraceState implements ProgressEventNotifier
{
public final UUID sessionId;
public final InetAddress coordinator;
public final Stopwatch watch;
public final ByteBuffer sessionIdBytes;
public final Tracing.TraceType traceType;
public final int ttl;
private boolean notify;
private final List<ProgressListener> listeners = new CopyOnWriteArrayList<>();
private String tag;
public enum Status
{
IDLE,
ACTIVE,
STOPPED
}
private volatile Status status;
private final AtomicInteger references = new AtomicInteger(1);
protected TraceState(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType)
{
assert coordinator != null;
assert sessionId != null;
this.coordinator = coordinator;
this.sessionId = sessionId;
sessionIdBytes = ByteBufferUtil.bytes(sessionId);
this.traceType = traceType;
this.ttl = traceType.getTTL();
watch = Stopwatch.createStarted();
this.status = Status.IDLE;
}
public void enableActivityNotification(String tag)
{
assert traceType == Tracing.TraceType.REPAIR;
notify = true;
this.tag = tag;
}
@Override
public void addProgressListener(ProgressListener listener)
{
assert traceType == Tracing.TraceType.REPAIR;
listeners.add(listener);
}
@Override
public void removeProgressListener(ProgressListener listener)
{
assert traceType == Tracing.TraceType.REPAIR;
listeners.remove(listener);
}
public int elapsed()
{
long elapsed = watch.elapsed(TimeUnit.MICROSECONDS);
return elapsed < Integer.MAX_VALUE ? (int) elapsed : Integer.MAX_VALUE;
}
public synchronized void stop()
{
waitForPendingEvents();
status = Status.STOPPED;
notifyAll();
}
public synchronized Status waitActivity(long timeout)
{
if (status == Status.IDLE)
{
try
{
wait(timeout);
}
catch (InterruptedException e)
{
throw new RuntimeException();
}
}
if (status == Status.ACTIVE)
{
status = Status.IDLE;
return Status.ACTIVE;
}
return status;
}
protected synchronized void notifyActivity()
{
status = Status.ACTIVE;
notifyAll();
}
public void trace(String format, Object arg)
{
trace(MessageFormatter.format(format, arg).getMessage());
}
public void trace(String format, Object arg1, Object arg2)
{
trace(MessageFormatter.format(format, arg1, arg2).getMessage());
}
public void trace(String format, Object[] args)
{
trace(MessageFormatter.arrayFormat(format, args).getMessage());
}
public void trace(String message)
{
if (notify)
notifyActivity();
traceImpl(message);
for (ProgressListener listener : listeners)
{
listener.progress(tag, ProgressEvent.createNotification(message));
}
}
protected abstract void traceImpl(String message);
protected void waitForPendingEvents()
{
}
public boolean acquireReference()
{
while (true)
{
int n = references.get();
if (n <= 0)
return false;
if (references.compareAndSet(n, n + 1))
return true;
}
}
public int releaseReference()
{
waitForPendingEvents();
return references.decrementAndGet();
}
}