wz2cool / local-queue / #31

Committed 01 Feb 2025 03:07AM UTC coverage: 90.052% (-1.8%) from 91.882%

Build # #31

Build Type

push

Committed by

Commit Message

add close

Coverage Stats

31 of 44 new or added lines in 4 files covered. (70.45%)

516 of 573 relevant lines covered (90.05%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.35

/src/main/java/com/github/wz2cool/localqueue/impl/SimpleConsumer.java

package com.github.wz2cool.localqueue.impl;

import com.github.wz2cool.localqueue.IConsumer;
import com.github.wz2cool.localqueue.event.CloseListener;
import com.github.wz2cool.localqueue.helper.ChronicleQueueHelper;
import com.github.wz2cool.localqueue.model.config.SimpleConsumerConfig;
import com.github.wz2cool.localqueue.model.enums.ConsumeFromWhere;
import com.github.wz2cool.localqueue.model.message.InternalReadMessage;
import com.github.wz2cool.localqueue.model.message.QueueMessage;
import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptTailer;
import net.openhft.chronicle.queue.RollCycle;
import net.openhft.chronicle.queue.RollCycles;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * simple consumer
 *
 * @author frank
 */
public class SimpleConsumer implements IConsumer {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private final RollCycle defaultRollCycle = RollCycles.FAST_DAILY;
    private final SimpleConsumerConfig config;
    private final PositionStore positionStore;
    private final SingleChronicleQueue queue;
    // should only call by readCacheExecutor
    private final ExcerptTailer mainTailer;
    private final ExecutorService readCacheExecutor = Executors.newSingleThreadExecutor();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private final LinkedBlockingQueue<QueueMessage> messageCache;
    private final ConcurrentLinkedQueue<CloseListener> closeListenerList = new ConcurrentLinkedQueue<>();
    private volatile long ackedReadPosition = -1;
    private volatile boolean isReadToCacheRunning = true;
    private volatile boolean isClosing = false;
    private volatile boolean isClosed = false;
    private final AtomicInteger positionVersion = new AtomicInteger(0);
    private final Lock internalLock = new ReentrantLock();

    /**
     * constructor
     *
     * @param config the config of consumer
     */
    public SimpleConsumer(final SimpleConsumerConfig config) {
        this.config = config;
        this.messageCache = new LinkedBlockingQueue<>(config.getCacheSize());
        this.positionStore = new PositionStore(config.getPositionFile());
        this.queue = ChronicleQueue.singleBuilder(config.getDataDir()).rollCycle(defaultRollCycle).build();
        this.mainTailer = initMainTailer();
        startReadToCache();
        scheduler.scheduleAtFixedRate(this::flushPosition, 0, config.getFlushPositionInterval(), TimeUnit.MILLISECONDS);
    }

    @Override
    public synchronized QueueMessage take() throws InterruptedException {
        return this.messageCache.take();
    }

    @Override
    public synchronized List<QueueMessage> batchTake(int maxBatchSize) throws InterruptedException {
        List<QueueMessage> result = new ArrayList<>(maxBatchSize);
        QueueMessage take = this.messageCache.take();
        result.add(take);
        this.messageCache.drainTo(result, maxBatchSize - 1);
        return result;
    }

    @Override
    public synchronized Optional<QueueMessage> take(long timeout, TimeUnit unit) throws InterruptedException {
        QueueMessage message = this.messageCache.poll(timeout, unit);
        return Optional.ofNullable(message);
    }

    @Override
    public synchronized List<QueueMessage> batchTake(int maxBatchSize, long timeout, TimeUnit unit) throws InterruptedException {
        List<QueueMessage> result = new ArrayList<>(maxBatchSize);
        QueueMessage poll = this.messageCache.poll(timeout, unit);
        if (Objects.nonNull(poll)) {
            result.add(poll);
            this.messageCache.drainTo(result, maxBatchSize - 1);
        }
        return result;
    }

    @Override
    public synchronized Optional<QueueMessage> poll() {
        QueueMessage message = this.messageCache.poll();
        return Optional.ofNullable(message);
    }

    @Override
    public synchronized List<QueueMessage> batchPoll(int maxBatchSize) {
        List<QueueMessage> result = new ArrayList<>(maxBatchSize);
        this.messageCache.drainTo(result, maxBatchSize);
        return result;
    }

    @Override
    public synchronized void ack(final QueueMessage message) {
        if (Objects.isNull(message)) {
            return;
        }

        if (message.getPositionVersion() != positionVersion.get()) {
            return;
        }

        this.ackedReadPosition = message.getPosition();
    }

    @Override
    public synchronized void ack(final List<QueueMessage> messages) {
        if (Objects.isNull(messages) || messages.isEmpty()) {
            return;
        }
        QueueMessage lastOne = messages.get(messages.size() - 1);
        if (lastOne.getPositionVersion() != positionVersion.get()) {
            return;
        }
        this.ackedReadPosition = lastOne.getPosition();
    }

    @Override
    public boolean moveToPosition(final long position) {
        logDebug("[moveToPosition] start");
        stopReadToCache();
        try {
            internalLock.lock();
            return moveToPositionInternal(position);
        } finally {
            internalLock.unlock();
            startReadToCache();
            logDebug("[moveToPosition] end");
        }
    }

    @Override
    public boolean moveToTimestamp(final long timestamp) {
        logDebug("[moveToTimestamp] start, timestamp: {}", timestamp);
        stopReadToCache();
        try {
            internalLock.lock();
            Optional<Long> positionOptional = findPosition(timestamp);
            if (!positionOptional.isPresent()) {
                return false;
            }
            Long position = positionOptional.get();
            return moveToPositionInternal(position);
        } finally {
            internalLock.unlock();
            startReadToCache();
            logDebug("[moveToTimestamp] end");
        }
    }

    @Override
    public Optional<QueueMessage> get(final String messageKey) {
        return get(messageKey, 0L, Long.MAX_VALUE);
    }

    @Override
    public Optional<QueueMessage> get(final String messageKey, long searchTimestampStart, long searchTimestampEnd) {
        if (messageKey == null || messageKey.isEmpty()) {
            return Optional.empty();
        }
        try (ExcerptTailer tailer = queue.createTailer()) {
            moveToNearByTimestamp(tailer, searchTimestampStart);
            while (true) {
                // for performance, ignore read content.
                InternalReadMessage internalReadMessage = new InternalReadMessage(true);
                boolean readResult = tailer.readBytes(internalReadMessage);
                if (!readResult) {
                    return Optional.empty();
                }
                if (internalReadMessage.getWriteTime() < searchTimestampStart) {
                    continue;
                }
                if (internalReadMessage.getWriteTime() > searchTimestampEnd) {
                    return Optional.empty();
                }
                boolean moveToResult = tailer.moveToIndex(tailer.lastReadIndex());
                if (!moveToResult) {
                    return Optional.empty();
                }
                internalReadMessage = new InternalReadMessage();
                readResult = tailer.readBytes(internalReadMessage);
                if (!readResult) {
                    return Optional.empty();
                }
                QueueMessage queueMessage = toQueueMessage(internalReadMessage, tailer.lastReadIndex());
                if (Objects.equals(messageKey, queueMessage.getMessageKey())) {
                    return Optional.of(queueMessage);
                }
            }
        }
    }

    private QueueMessage toQueueMessage(final InternalReadMessage internalReadMessage, final long position) {
        return new QueueMessage(
                internalReadMessage.getMessageKey(),
                positionVersion.get(),
                position,
                internalReadMessage.getContent(),
                internalReadMessage.getWriteTime());
    }

    private boolean moveToPositionInternal(final long position) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                logDebug("[moveToPositionInternal] start, position: {}", position);
                boolean moveToResult = mainTailer.moveToIndex(position);
                if (moveToResult) {
                    positionVersion.incrementAndGet();
                    messageCache.clear();
                    this.ackedReadPosition = position;
                }
                return moveToResult;
            } finally {
                logDebug("[moveToPositionInternal] end");
            }
        }, this.readCacheExecutor).join();
    }

    private Optional<Long> findPosition(final long timestamp) {
        logDebug("[findPosition] start, timestamp: {}", timestamp);
        try (ExcerptTailer tailer = queue.createTailer()) {
            moveToNearByTimestamp(tailer, timestamp);
            while (true) {
                InternalReadMessage internalReadMessage = new InternalReadMessage(true);
                boolean resultResult = tailer.readBytes(internalReadMessage);
                if (resultResult) {
                    if (internalReadMessage.getWriteTime() >= timestamp) {
                        return Optional.of(tailer.lastReadIndex());
                    }
                } else {
                    return Optional.empty();
                }
            }
        } finally {
            logDebug("[findPosition] end");
        }
    }

    public long getAckedReadPosition() {
        return ackedReadPosition;
    }

    public boolean isClosed() {
        return isClosed;
    }

    private void stopReadToCache() {
        this.isReadToCacheRunning = false;
    }

    private void startReadToCache() {
        this.isReadToCacheRunning = true;
        readCacheExecutor.execute(this::readToCache);
    }

    private void readToCache() {
        try {
            logDebug("[readToCache] start");
            internalLock.lock();
            long pullInterval = config.getPullInterval();
            long fillCacheInterval = config.getFillCacheInterval();
            while (this.isReadToCacheRunning && !isClosing) {
                try {
                    InternalReadMessage internalReadMessage = new InternalReadMessage();
                    boolean readResult = mainTailer.readBytes(internalReadMessage);
                    if (!readResult) {
                        TimeUnit.MILLISECONDS.sleep(pullInterval);
                        continue;
                    }
                    long lastedReadIndex = mainTailer.lastReadIndex();
                    QueueMessage queueMessage = toQueueMessage(internalReadMessage, lastedReadIndex);
                    boolean offerResult = this.messageCache.offer(queueMessage, fillCacheInterval, TimeUnit.MILLISECONDS);
                    if (!offerResult) {
                        // if offer failed, move to last read position
                        mainTailer.moveToIndex(lastedReadIndex);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } finally {
            internalLock.unlock();
            logDebug("[readToCache] end");
        }
    }

    private ExcerptTailer initMainTailer() {
        return CompletableFuture.supplyAsync(this::initMainTailerInternal, this.readCacheExecutor).join();
    }

    private ExcerptTailer initMainTailerInternal() {
        try {
            logDebug("[initExcerptTailerInternal] start");
            ExcerptTailer tailer = queue.createTailer();
            Optional<Long> lastPositionOptional = getLastPosition();
            if (lastPositionOptional.isPresent()) {
                Long position = lastPositionOptional.get();
                long beginPosition = position + 1;
                tailer.moveToIndex(beginPosition);
                logDebug("[initExcerptTailerInternal] find last position and move to position: {}", beginPosition);
            } else {
                ConsumeFromWhere consumeFromWhere = this.config.getConsumeFromWhere();
                if (consumeFromWhere == ConsumeFromWhere.LAST) {
                    tailer.toEnd();
                    logDebug("[initExcerptTailerInternal] move to end");
                } else if (consumeFromWhere == ConsumeFromWhere.FIRST) {
                    tailer.toStart();
                    logDebug("[initExcerptTailerInternal] move to start");
                }
            }
            return tailer;
        } finally {
            logDebug("[initExcerptTailer] end");
        }

    }

    /// region position

    private void flushPosition() {
        if (ackedReadPosition != -1) {
            setLastPosition(this.ackedReadPosition);
        }
    }

    private Optional<Long> getLastPosition() {
        Long position = positionStore.get(config.getConsumerId());
        if (position == null) {
            return Optional.empty();
        }
        return Optional.of(position);
    }

    private void setLastPosition(long position) {
        positionStore.put(config.getConsumerId(), position);
    }

    /// endregion

    @Override
    public void close() {
        try {
            logDebug("[close] start");
            if (isClosed) {
                logDebug("[close] already closed");
                return;
            }
            isClosing = true;
            this.internalLock.lock();
            stopReadToCache();
            this.internalLock.unlock();
            if (!positionStore.isClosed()) {
                positionStore.close();
            }
            scheduler.shutdown();
            readCacheExecutor.shutdown();
            try {
                if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
                if (!readCacheExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
                    readCacheExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduler.shutdownNow();
                readCacheExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
            if (!queue.isClosed()) {
                queue.close();
            }

            for (CloseListener closeListener : closeListenerList) {
                closeListener.onClose();
            }
            isClosed = true;
        } finally {
            logDebug("[close] end");
        }
    }

    private void moveToNearByTimestamp(ExcerptTailer tailer, long timestamp) {
        int expectedCycle = ChronicleQueueHelper.cycle(defaultRollCycle, timestamp);
        int currentCycle = tailer.cycle();
        if (currentCycle != expectedCycle) {
            boolean moveToCycleResult = tailer.moveToCycle(expectedCycle);
            logDebug("[moveToNearByTimestamp] moveToCycleResult: {}", moveToCycleResult);
        }
    }

    @Override
    public void addCloseListener(CloseListener listener) {
        closeListenerList.add(listener);
    }

    // region logger

    private void logDebug(String format) {
        if (logger.isDebugEnabled()) {
            logger.debug(format);
        }
    }

    private void logDebug(String format, Object arg) {
        if (logger.isDebugEnabled()) {
            logger.debug(format, arg);
        }
    }

    // endregion
}

1	package com.github.wz2cool.localqueue.impl;
2
3	import com.github.wz2cool.localqueue.IConsumer;
4	import com.github.wz2cool.localqueue.event.CloseListener;
5	import com.github.wz2cool.localqueue.helper.ChronicleQueueHelper;
6	import com.github.wz2cool.localqueue.model.config.SimpleConsumerConfig;
7	import com.github.wz2cool.localqueue.model.enums.ConsumeFromWhere;
8	import com.github.wz2cool.localqueue.model.message.InternalReadMessage;
9	import com.github.wz2cool.localqueue.model.message.QueueMessage;
10	import net.openhft.chronicle.queue.ChronicleQueue;
11	import net.openhft.chronicle.queue.ExcerptTailer;
12	import net.openhft.chronicle.queue.RollCycle;
13	import net.openhft.chronicle.queue.RollCycles;
14	import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
15	import org.slf4j.Logger;
16	import org.slf4j.LoggerFactory;
17
18	import java.util.ArrayList;
19	import java.util.List;
20	import java.util.Objects;
21	import java.util.Optional;
22	import java.util.concurrent.*;
23	import java.util.concurrent.atomic.AtomicInteger;
24	import java.util.concurrent.locks.Lock;
25	import java.util.concurrent.locks.ReentrantLock;
26
27	/**
28	* simple consumer
29	*
30	* @author frank
31	*/
32	public class SimpleConsumer implements IConsumer {
33
34	private final Logger logger = LoggerFactory.getLogger(this.getClass());	1✔
35	private final RollCycle defaultRollCycle = RollCycles.FAST_DAILY;	1✔
36	private final SimpleConsumerConfig config;
37	private final PositionStore positionStore;
38	private final SingleChronicleQueue queue;
39	// should only call by readCacheExecutor
40	private final ExcerptTailer mainTailer;
41	private final ExecutorService readCacheExecutor = Executors.newSingleThreadExecutor();	1✔
42	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);	1✔
43	private final LinkedBlockingQueue<QueueMessage> messageCache;
44	private final ConcurrentLinkedQueue<CloseListener> closeListenerList = new ConcurrentLinkedQueue<>();	1✔
45	private volatile long ackedReadPosition = -1;	1✔
46	private volatile boolean isReadToCacheRunning = true;	1✔
47	private volatile boolean isClosing = false;	1✔
48	private volatile boolean isClosed = false;	1✔
49	private final AtomicInteger positionVersion = new AtomicInteger(0);	1✔
50	private final Lock internalLock = new ReentrantLock();	1✔
51
52	/**
53	* constructor
54	*
55	* @param config the config of consumer
56	*/
57	public SimpleConsumer(final SimpleConsumerConfig config) {	1✔
58	this.config = config;	1✔
59	this.messageCache = new LinkedBlockingQueue<>(config.getCacheSize());	1✔
60	this.positionStore = new PositionStore(config.getPositionFile());	1✔
61	this.queue = ChronicleQueue.singleBuilder(config.getDataDir()).rollCycle(defaultRollCycle).build();	1✔
62	this.mainTailer = initMainTailer();	1✔
63	startReadToCache();	1✔
64	scheduler.scheduleAtFixedRate(this::flushPosition, 0, config.getFlushPositionInterval(), TimeUnit.MILLISECONDS);	1✔
65	}	1✔
66
67	@Override
68	public synchronized QueueMessage take() throws InterruptedException {
69	return this.messageCache.take();	1✔
70	}
71
72	@Override
73	public synchronized List<QueueMessage> batchTake(int maxBatchSize) throws InterruptedException {
74	List<QueueMessage> result = new ArrayList<>(maxBatchSize);	1✔
75	QueueMessage take = this.messageCache.take();	1✔
76	result.add(take);	1✔
77	this.messageCache.drainTo(result, maxBatchSize - 1);	1✔
78	return result;	1✔
79	}
80
81	@Override
82	public synchronized Optional<QueueMessage> take(long timeout, TimeUnit unit) throws InterruptedException {
83	QueueMessage message = this.messageCache.poll(timeout, unit);	1✔
84	return Optional.ofNullable(message);	1✔
85	}
86
87	@Override
88	public synchronized List<QueueMessage> batchTake(int maxBatchSize, long timeout, TimeUnit unit) throws InterruptedException {
89	List<QueueMessage> result = new ArrayList<>(maxBatchSize);	1✔
90	QueueMessage poll = this.messageCache.poll(timeout, unit);	1✔
91	if (Objects.nonNull(poll)) {	1✔
92	result.add(poll);	1✔
93	this.messageCache.drainTo(result, maxBatchSize - 1);	1✔
94	}
95	return result;	1✔
96	}
97
98	@Override
99	public synchronized Optional<QueueMessage> poll() {
100	QueueMessage message = this.messageCache.poll();	1✔
101	return Optional.ofNullable(message);	1✔
102	}
103
104	@Override
105	public synchronized List<QueueMessage> batchPoll(int maxBatchSize) {
106	List<QueueMessage> result = new ArrayList<>(maxBatchSize);	1✔
107	this.messageCache.drainTo(result, maxBatchSize);	1✔
108	return result;	1✔
109	}
110
111	@Override
112	public synchronized void ack(final QueueMessage message) {
113	if (Objects.isNull(message)) {	1✔
114	return;	1✔
115	}
116
117	if (message.getPositionVersion() != positionVersion.get()) {	1✔
118	return;	×
119	}
120
121	this.ackedReadPosition = message.getPosition();	1✔
122	}	1✔
123
124	@Override
125	public synchronized void ack(final List<QueueMessage> messages) {
126	if (Objects.isNull(messages) \|\| messages.isEmpty()) {	1✔
127	return;	1✔
128	}
129	QueueMessage lastOne = messages.get(messages.size() - 1);	1✔
130	if (lastOne.getPositionVersion() != positionVersion.get()) {	1✔
131	return;	×
132	}
133	this.ackedReadPosition = lastOne.getPosition();	1✔
134	}	1✔
135
136	@Override
137	public boolean moveToPosition(final long position) {
138	logDebug("[moveToPosition] start");	1✔
139	stopReadToCache();	1✔
140	try {
141	internalLock.lock();	1✔
142	return moveToPositionInternal(position);	1✔
143	} finally {
144	internalLock.unlock();	1✔
145	startReadToCache();	1✔
146	logDebug("[moveToPosition] end");	1✔
147	}	×
148	}
149
150	@Override
151	public boolean moveToTimestamp(final long timestamp) {
152	logDebug("[moveToTimestamp] start, timestamp: {}", timestamp);	1✔
153	stopReadToCache();	1✔
154	try {
155	internalLock.lock();	1✔
156	Optional<Long> positionOptional = findPosition(timestamp);	1✔
157	if (!positionOptional.isPresent()) {	1✔
158	return false;	1✔
159	}
160	Long position = positionOptional.get();	1✔
161	return moveToPositionInternal(position);	1✔
162	} finally {
163	internalLock.unlock();	1✔
164	startReadToCache();	1✔
165	logDebug("[moveToTimestamp] end");	1✔
166	}	×
167	}
168
169	@Override
170	public Optional<QueueMessage> get(final String messageKey) {
171	return get(messageKey, 0L, Long.MAX_VALUE);	1✔
172	}
173
174	@Override
175	public Optional<QueueMessage> get(final String messageKey, long searchTimestampStart, long searchTimestampEnd) {
176	if (messageKey == null \|\| messageKey.isEmpty()) {	1✔
177	return Optional.empty();	1✔
178	}
179	try (ExcerptTailer tailer = queue.createTailer()) {	1✔
180	moveToNearByTimestamp(tailer, searchTimestampStart);	1✔
181	while (true) {
182	// for performance, ignore read content.
183	InternalReadMessage internalReadMessage = new InternalReadMessage(true);	1✔
184	boolean readResult = tailer.readBytes(internalReadMessage);	1✔
185	if (!readResult) {	1✔
186	return Optional.empty();	1✔
187	}
188	if (internalReadMessage.getWriteTime() < searchTimestampStart) {	1✔
189	continue;	1✔
190	}
191	if (internalReadMessage.getWriteTime() > searchTimestampEnd) {	1✔
192	return Optional.empty();	×
193	}
194	boolean moveToResult = tailer.moveToIndex(tailer.lastReadIndex());	1✔
195	if (!moveToResult) {	1✔
196	return Optional.empty();	×
197	}
198	internalReadMessage = new InternalReadMessage();	1✔
199	readResult = tailer.readBytes(internalReadMessage);	1✔
200	if (!readResult) {	1✔
201	return Optional.empty();	×
202	}
203	QueueMessage queueMessage = toQueueMessage(internalReadMessage, tailer.lastReadIndex());	1✔
204	if (Objects.equals(messageKey, queueMessage.getMessageKey())) {	1✔
205	return Optional.of(queueMessage);	1✔
206	}
207	}	1✔
208	}	1✔
209	}
210
211	private QueueMessage toQueueMessage(final InternalReadMessage internalReadMessage, final long position) {
212	return new QueueMessage(	1✔
213	internalReadMessage.getMessageKey(),	1✔
214	positionVersion.get(),	1✔
215	position,
216	internalReadMessage.getContent(),	1✔
217	internalReadMessage.getWriteTime());	1✔
218	}
219
220	private boolean moveToPositionInternal(final long position) {
221	return CompletableFuture.supplyAsync(() -> {	1✔
222	try {
223	logDebug("[moveToPositionInternal] start, position: {}", position);	1✔
224	boolean moveToResult = mainTailer.moveToIndex(position);	1✔
225	if (moveToResult) {	1✔
226	positionVersion.incrementAndGet();	1✔
227	messageCache.clear();	1✔
228	this.ackedReadPosition = position;	1✔
229	}
230	return moveToResult;	1✔
231	} finally {
232	logDebug("[moveToPositionInternal] end");	1✔
233	}	×
234	}, this.readCacheExecutor).join();	1✔
235	}
236
237	private Optional<Long> findPosition(final long timestamp) {
238	logDebug("[findPosition] start, timestamp: {}", timestamp);	1✔
239	try (ExcerptTailer tailer = queue.createTailer()) {	1✔
240	moveToNearByTimestamp(tailer, timestamp);	1✔
241	while (true) {
242	InternalReadMessage internalReadMessage = new InternalReadMessage(true);	1✔
243	boolean resultResult = tailer.readBytes(internalReadMessage);	1✔
244	if (resultResult) {	1✔
245	if (internalReadMessage.getWriteTime() >= timestamp) {	1✔
246	return Optional.of(tailer.lastReadIndex());	1✔
247	}
248	} else {
249	return Optional.empty();	1✔
250	}
251	}	1✔
252	} finally {	1✔
253	logDebug("[findPosition] end");	1✔
254	}	×
255	}
256
257	public long getAckedReadPosition() {
258	return ackedReadPosition;	1✔
259	}
260
261	public boolean isClosed() {
262	return isClosed;	1✔
263	}
264
265	private void stopReadToCache() {
266	this.isReadToCacheRunning = false;	1✔
267	}	1✔
268
269	private void startReadToCache() {
270	this.isReadToCacheRunning = true;	1✔
271	readCacheExecutor.execute(this::readToCache);	1✔
272	}	1✔
273
274	private void readToCache() {
275	try {
276	logDebug("[readToCache] start");	1✔
277	internalLock.lock();	1✔
278	long pullInterval = config.getPullInterval();	1✔
279	long fillCacheInterval = config.getFillCacheInterval();	1✔
280	while (this.isReadToCacheRunning && !isClosing) {	1✔
281	try {
282	InternalReadMessage internalReadMessage = new InternalReadMessage();	1✔
283	boolean readResult = mainTailer.readBytes(internalReadMessage);	1✔
284	if (!readResult) {	1✔
285	TimeUnit.MILLISECONDS.sleep(pullInterval);	1✔
286	continue;	1✔
287	}
288	long lastedReadIndex = mainTailer.lastReadIndex();	1✔
289	QueueMessage queueMessage = toQueueMessage(internalReadMessage, lastedReadIndex);	1✔
290	boolean offerResult = this.messageCache.offer(queueMessage, fillCacheInterval, TimeUnit.MILLISECONDS);	1✔
291	if (!offerResult) {	1✔
292	// if offer failed, move to last read position
293	mainTailer.moveToIndex(lastedReadIndex);	1✔
294	}
295	} catch (InterruptedException e) {	×
296	Thread.currentThread().interrupt();	×
297	}	1✔
298	}
299	} finally {
300	internalLock.unlock();	1✔
301	logDebug("[readToCache] end");	1✔
302	}	1✔
303	}	1✔
304
305	private ExcerptTailer initMainTailer() {
306	return CompletableFuture.supplyAsync(this::initMainTailerInternal, this.readCacheExecutor).join();	1✔
307	}
308
309	private ExcerptTailer initMainTailerInternal() {
310	try {
311	logDebug("[initExcerptTailerInternal] start");	1✔
312	ExcerptTailer tailer = queue.createTailer();	1✔
313	Optional<Long> lastPositionOptional = getLastPosition();	1✔
314	if (lastPositionOptional.isPresent()) {	1✔
315	Long position = lastPositionOptional.get();	1✔
316	long beginPosition = position + 1;	1✔
317	tailer.moveToIndex(beginPosition);	1✔
318	logDebug("[initExcerptTailerInternal] find last position and move to position: {}", beginPosition);	1✔
319	} else {	1✔
320	ConsumeFromWhere consumeFromWhere = this.config.getConsumeFromWhere();	1✔
321	if (consumeFromWhere == ConsumeFromWhere.LAST) {	1✔
322	tailer.toEnd();	1✔
323	logDebug("[initExcerptTailerInternal] move to end");	1✔
324	} else if (consumeFromWhere == ConsumeFromWhere.FIRST) {	1✔
325	tailer.toStart();	1✔
326	logDebug("[initExcerptTailerInternal] move to start");	1✔
327	}
328	}
329	return tailer;	1✔
330	} finally {
331	logDebug("[initExcerptTailer] end");	1✔
332	}	×
333
334	}
335
336	/// region position
337
338	private void flushPosition() {
339	if (ackedReadPosition != -1) {	1✔
340	setLastPosition(this.ackedReadPosition);	1✔
341	}
342	}	1✔
343
344	private Optional<Long> getLastPosition() {
345	Long position = positionStore.get(config.getConsumerId());	1✔
346	if (position == null) {	1✔
347	return Optional.empty();	1✔
348	}
349	return Optional.of(position);	1✔
350	}
351
352	private void setLastPosition(long position) {
353	positionStore.put(config.getConsumerId(), position);	1✔
354	}	1✔
355
356	/// endregion
357
358	@Override
359	public void close() {
360	try {
361	logDebug("[close] start");	1✔
362	if (isClosed) {	1✔
NEW 363	logDebug("[close] already closed");	×
NEW 364	return;	×
365	}
366	isClosing = true;	1✔
367	this.internalLock.lock();	1✔
368	stopReadToCache();	1✔
369	this.internalLock.unlock();	1✔
370	if (!positionStore.isClosed()) {	1✔
371	positionStore.close();	1✔
372	}
373	scheduler.shutdown();	1✔
374	readCacheExecutor.shutdown();	1✔
375	try {
376	if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
377	scheduler.shutdownNow();	×
378	}
379	if (!readCacheExecutor.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
380	readCacheExecutor.shutdownNow();	×
381	}
382	} catch (InterruptedException e) {	×
383	scheduler.shutdownNow();	×
384	readCacheExecutor.shutdownNow();	×
385	Thread.currentThread().interrupt();	×
386	}	1✔
387	if (!queue.isClosed()) {	1✔
388	queue.close();	1✔
389	}
390
391	for (CloseListener closeListener : closeListenerList) {	1✔
392	closeListener.onClose();	1✔
393	}	1✔
394	isClosed = true;	1✔
395	} finally {
396	logDebug("[close] end");	1✔
397	}	1✔
398	}	1✔
399
400	private void moveToNearByTimestamp(ExcerptTailer tailer, long timestamp) {
401	int expectedCycle = ChronicleQueueHelper.cycle(defaultRollCycle, timestamp);	1✔
402	int currentCycle = tailer.cycle();	1✔
403	if (currentCycle != expectedCycle) {	1✔
404	boolean moveToCycleResult = tailer.moveToCycle(expectedCycle);	1✔
405	logDebug("[moveToNearByTimestamp] moveToCycleResult: {}", moveToCycleResult);	1✔
406	}
407	}	1✔
408
409	@Override
410	public void addCloseListener(CloseListener listener) {
411	closeListenerList.add(listener);	1✔
412	}	1✔
413
414	// region logger
415
416	private void logDebug(String format) {
417	if (logger.isDebugEnabled()) {	1✔
418	logger.debug(format);	×
419	}
420	}	1✔
421
422	private void logDebug(String format, Object arg) {
423	if (logger.isDebugEnabled()) {	1✔
424	logger.debug(format, arg);	×
425	}
426	}	1✔
427
428	// endregion
429	}

wz2cool / local-queue / #31

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous