wz2cool / local-queue / #48

Committed 03 Feb 2025 12:40PM UTC coverage: 91.935% (+0.09%) from 91.85%

Build # #48

Build Type

push

Committed by

Commit Message

Merge pull request #7 from wz2cool/0.1.3

0.1.3

Run Details

221 of 248 new or added lines in 7 files covered. (89.11%)

5 existing lines in 2 files now uncovered.

684 of 744 relevant lines covered (91.94%)

0.92 hits per line

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

91.06

/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 com.github.wz2cool.localqueue.model.page.PageInfo;
import com.github.wz2cool.localqueue.model.page.SortDirection;
import com.github.wz2cool.localqueue.model.page.UpDown;
import net.openhft.chronicle.core.time.TimeProvider;
import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptTailer;
import net.openhft.chronicle.queue.RollCycle;
import net.openhft.chronicle.queue.TailerDirection;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

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

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private final RollCycle defaultRollCycle;
    private final TimeProvider timeProvider;
    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 final AtomicLong ackedReadPosition = new AtomicLong(-1);
    private final AtomicBoolean isReadToCacheRunning = new AtomicBoolean(true);
    private final AtomicBoolean isClosing = new AtomicBoolean(false);
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final Object closeLocker = new Object();
    private final AtomicInteger positionVersion = new AtomicInteger(0);

    /**
     * constructor
     *
     * @param config the config of consumer
     */
    public SimpleConsumer(final SimpleConsumerConfig config) {
        this.config = config;
        this.timeProvider = ChronicleQueueHelper.getTimeProvider(config.getTimeZone());
        this.messageCache = new LinkedBlockingQueue<>(config.getCacheSize());
        this.positionStore = new PositionStore(config.getPositionFile());
        this.defaultRollCycle = ChronicleQueueHelper.getRollCycle(config.getRollCycleType());
        this.queue = ChronicleQueue.singleBuilder(config.getDataDir())
                .timeProvider(timeProvider)
                .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;
        }
        ackedReadPosition.set(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;
        }
        ackedReadPosition.set(lastOne.getPosition());
    }

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

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

    @Override
    public Optional<QueueMessage> get(final long position) {
        if (position < 0) {
            return Optional.empty();
        }
        try (ExcerptTailer tailer = queue.createTailer()) {
            tailer.moveToIndex(position);
            InternalReadMessage internalReadMessage = new InternalReadMessage();
            boolean readResult = tailer.readBytes(internalReadMessage);
            if (readResult) {
                return Optional.of(toQueueMessage(internalReadMessage, position));
            } else {
                return Optional.empty();
            }
        }
    }

    @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(() -> {
            synchronized (closeLocker) {
                try {
                    if (isClosing.get()) {
                        logDebug("[moveToPositionInternal] consumer is closing");
                        return false;
                    }
                    logDebug("[moveToPositionInternal] start, position: {}", position);
                    boolean moveToResult = mainTailer.moveToIndex(position);
                    if (moveToResult) {
                        positionVersion.incrementAndGet();
                        messageCache.clear();
                        ackedReadPosition.set(position);
                    }
                    return moveToResult;
                } finally {
                    logDebug("[moveToPositionInternal] end");
                }
            }
        }, this.readCacheExecutor).join();
    }


    @Override
    public 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.get();
    }

    @Override
    public boolean isClosed() {
        return isClosed.get();
    }

    private void stopReadToCache() {
        isReadToCacheRunning.set(false);
    }

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

    private void readToCache() {
        try {
            logDebug("[readToCache] start");
            long pullInterval = config.getPullInterval();
            long fillCacheInterval = config.getFillCacheInterval();
            while (isReadToCacheRunning.get() && !isClosing.get()) {
                synchronized (closeLocker) {
                    try {
                        if (isClosing.get()) {
                            logDebug("[readToCache] consumer is closing");
                            return;
                        }
                        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 {
            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.get() != -1) {
            setLastPosition(this.ackedReadPosition.get());
        }
    }

    private Optional<Long> getLastPosition() {
        return positionStore.get(config.getConsumerId());
    }

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

    /// endregion

    @SuppressWarnings("Duplicates")
    @Override
    public void close() {
        synchronized (closeLocker) {
            try {
                logDebug("[close] start");
                if (isClosing.get()) {
                    logDebug("[close] is closing");
                    return;
                }
                isClosing.set(true);
                stopReadToCache();
                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.set(true);
            } finally {
                logDebug("[close] end");
            }
        }
    }

    private void moveToNearByTimestamp(ExcerptTailer tailer, long timestamp) {
        int expectedCycle = ChronicleQueueHelper.cycle(defaultRollCycle, timeProvider, 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 page

    @Override
    public PageInfo<QueueMessage> getPage(SortDirection sortDirection, int pageSize) {
        return getPage(-1, sortDirection, pageSize);
    }

    @SuppressWarnings("Duplicates")
    @Override
    public PageInfo<QueueMessage> getPage(long moveToPosition, SortDirection sortDirection, int pageSize) {
        try (ExcerptTailer tailer = queue.createTailer()) {
            if (moveToPosition != -1) {
                tailer.moveToIndex(moveToPosition);
            }
            if (sortDirection == SortDirection.DESC) {
                tailer.toEnd();
                tailer.direction(TailerDirection.BACKWARD);
            }
            List<QueueMessage> data = new ArrayList<>();
            long start = -1;
            long end = -1;
            for (int i = 0; i < pageSize; i++) {
                InternalReadMessage internalReadMessage = new InternalReadMessage();
                boolean readResult = tailer.readBytes(internalReadMessage);
                if (!readResult) {
                    break;
                }
                QueueMessage queueMessage = toQueueMessage(internalReadMessage, tailer.lastReadIndex());
                data.add(queueMessage);
                if (i == 0) {
                    start = tailer.lastReadIndex();
                }
                end = tailer.lastReadIndex();
            }
            return new PageInfo<>(start, end, data, sortDirection, pageSize);
        }
    }

    @SuppressWarnings("Duplicates")
    @Override
    public PageInfo<QueueMessage> getPage(PageInfo<QueueMessage> prevPageInfo, UpDown upDown) {
        SortDirection sortDirection = prevPageInfo.getSortDirection();
        int pageSize = prevPageInfo.getPageSize();
        long start = prevPageInfo.getStart();
        long end = prevPageInfo.getEnd();
        try (ExcerptTailer tailer = queue.createTailer()) {
            TailerDirection tailerDirection = getTailerDirection(sortDirection, upDown);
            tailer.direction(tailerDirection);
            if (sortDirection == SortDirection.DESC) {
                if (upDown == UpDown.DOWN) {
                    tailer.moveToIndex(end - 1);
                } else {
                    tailer.moveToIndex(start + 1);
                }
            } else {
                if (upDown == UpDown.DOWN) {
                    tailer.moveToIndex(end + 1);
                } else {
                    tailer.moveToIndex(start - 1);
                }
            }
            List<QueueMessage> data = new ArrayList<>();
            for (int i = 0; i < pageSize; i++) {
                InternalReadMessage internalReadMessage = new InternalReadMessage();
                boolean readResult = tailer.readBytes(internalReadMessage);
                if (!readResult) {
                    break;
                }
                QueueMessage queueMessage = toQueueMessage(internalReadMessage, tailer.lastReadIndex());
                data.add(queueMessage);
                if (i == 0) {
                    start = tailer.lastReadIndex();
                }
                end = tailer.lastReadIndex();
            }
            if (upDown == UpDown.UP) {
                Collections.reverse(data);
            }
            return new PageInfo<>(start, end, data, sortDirection, pageSize);
        }
    }

    private TailerDirection getTailerDirection(SortDirection sortDirection, UpDown upDown) {
        if (sortDirection == SortDirection.DESC && upDown == UpDown.DOWN) {
            return TailerDirection.BACKWARD;
        }
        if (sortDirection == SortDirection.DESC && upDown == UpDown.UP) {
            return TailerDirection.FORWARD;
        }
        if (sortDirection == SortDirection.ASC && upDown == UpDown.DOWN) {
            return TailerDirection.FORWARD;
        }
        if (sortDirection == SortDirection.ASC && upDown == UpDown.UP) {
            return TailerDirection.BACKWARD;
        }
        return TailerDirection.FORWARD;
    }


    // endregion

    // 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 com.github.wz2cool.localqueue.model.page.PageInfo;
11	import com.github.wz2cool.localqueue.model.page.SortDirection;
12	import com.github.wz2cool.localqueue.model.page.UpDown;
13	import net.openhft.chronicle.core.time.TimeProvider;
14	import net.openhft.chronicle.queue.ChronicleQueue;
15	import net.openhft.chronicle.queue.ExcerptTailer;
16	import net.openhft.chronicle.queue.RollCycle;
17	import net.openhft.chronicle.queue.TailerDirection;
18	import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
19	import org.slf4j.Logger;
20	import org.slf4j.LoggerFactory;
21
22	import java.util.*;
23	import java.util.concurrent.*;
24	import java.util.concurrent.atomic.AtomicBoolean;
25	import java.util.concurrent.atomic.AtomicInteger;
26	import java.util.concurrent.atomic.AtomicLong;
27
28	/**
29	* simple consumer
30	*
31	* @author frank
32	*/
33	public class SimpleConsumer implements IConsumer {
34
35	private final Logger logger = LoggerFactory.getLogger(this.getClass());	1✔
36	private final RollCycle defaultRollCycle;
37	private final TimeProvider timeProvider;
38	private final SimpleConsumerConfig config;
39	private final PositionStore positionStore;
40	private final SingleChronicleQueue queue;
41	// should only call by readCacheExecutor
42	private final ExcerptTailer mainTailer;
43	private final ExecutorService readCacheExecutor = Executors.newSingleThreadExecutor();	1✔
44	private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);	1✔
45	private final LinkedBlockingQueue<QueueMessage> messageCache;
46	private final ConcurrentLinkedQueue<CloseListener> closeListenerList = new ConcurrentLinkedQueue<>();	1✔
47	private final AtomicLong ackedReadPosition = new AtomicLong(-1);	1✔
48	private final AtomicBoolean isReadToCacheRunning = new AtomicBoolean(true);	1✔
49	private final AtomicBoolean isClosing = new AtomicBoolean(false);	1✔
50	private final AtomicBoolean isClosed = new AtomicBoolean(false);	1✔
51	private final Object closeLocker = new Object();	1✔
52	private final AtomicInteger positionVersion = new AtomicInteger(0);	1✔
53
54	/**
55	* constructor
56	*
57	* @param config the config of consumer
58	*/
59	public SimpleConsumer(final SimpleConsumerConfig config) {	1✔
60	this.config = config;	1✔
61	this.timeProvider = ChronicleQueueHelper.getTimeProvider(config.getTimeZone());	1✔
62	this.messageCache = new LinkedBlockingQueue<>(config.getCacheSize());	1✔
63	this.positionStore = new PositionStore(config.getPositionFile());	1✔
64	this.defaultRollCycle = ChronicleQueueHelper.getRollCycle(config.getRollCycleType());	1✔
65	this.queue = ChronicleQueue.singleBuilder(config.getDataDir())	1✔
66	.timeProvider(timeProvider)	1✔
67	.rollCycle(defaultRollCycle)	1✔
68	.build();	1✔
69	this.mainTailer = initMainTailer();	1✔
70	startReadToCache();	1✔
71	scheduler.scheduleAtFixedRate(this::flushPosition, 0, config.getFlushPositionInterval(), TimeUnit.MILLISECONDS);	1✔
72	}	1✔
73
74	@Override
75	public synchronized QueueMessage take() throws InterruptedException {
76	return this.messageCache.take();	1✔
77	}
78
79	@Override
80	public synchronized List<QueueMessage> batchTake(int maxBatchSize) throws InterruptedException {
81	List<QueueMessage> result = new ArrayList<>(maxBatchSize);	1✔
82	QueueMessage take = this.messageCache.take();	1✔
83	result.add(take);	1✔
84	this.messageCache.drainTo(result, maxBatchSize - 1);	1✔
85	return result;	1✔
86	}
87
88	@Override
89	public synchronized Optional<QueueMessage> take(long timeout, TimeUnit unit) throws InterruptedException {
90	QueueMessage message = this.messageCache.poll(timeout, unit);	1✔
91	return Optional.ofNullable(message);	1✔
92	}
93
94	@Override
95	public synchronized List<QueueMessage> batchTake(int maxBatchSize, long timeout, TimeUnit unit) throws InterruptedException {
96	List<QueueMessage> result = new ArrayList<>(maxBatchSize);	1✔
97	QueueMessage poll = this.messageCache.poll(timeout, unit);	1✔
98	if (Objects.nonNull(poll)) {	1✔
99	result.add(poll);	1✔
100	this.messageCache.drainTo(result, maxBatchSize - 1);	1✔
101	}
102	return result;	1✔
103	}
104
105	@Override
106	public synchronized Optional<QueueMessage> poll() {
107	QueueMessage message = this.messageCache.poll();	1✔
108	return Optional.ofNullable(message);	1✔
109	}
110
111	@Override
112	public synchronized List<QueueMessage> batchPoll(int maxBatchSize) {
113	List<QueueMessage> result = new ArrayList<>(maxBatchSize);	1✔
114	this.messageCache.drainTo(result, maxBatchSize);	1✔
115	return result;	1✔
116	}
117
118	@Override
119	public synchronized void ack(final QueueMessage message) {
120	if (Objects.isNull(message)) {	1✔
121	return;	1✔
122	}
123
124	if (message.getPositionVersion() != positionVersion.get()) {	1✔
125	return;	×
126	}
127	ackedReadPosition.set(message.getPosition());	1✔
128	}	1✔
129
130	@Override
131	public synchronized void ack(final List<QueueMessage> messages) {
132	if (Objects.isNull(messages) \|\| messages.isEmpty()) {	1✔
133	return;	1✔
134	}
135	QueueMessage lastOne = messages.get(messages.size() - 1);	1✔
136	if (lastOne.getPositionVersion() != positionVersion.get()) {	1✔
137	return;	×
138	}
139	ackedReadPosition.set(lastOne.getPosition());	1✔
140	}	1✔
141
142	@Override
143	public boolean moveToPosition(final long position) {
144	logDebug("[moveToPosition] start");	1✔
145	stopReadToCache();	1✔
146	try {
147	return moveToPositionInternal(position);	1✔
148	} finally {
149	startReadToCache();	1✔
150	logDebug("[moveToPosition] end");	1✔
151	}	×
152	}
153
154	@Override
155	public boolean moveToTimestamp(final long timestamp) {
156	logDebug("[moveToTimestamp] start, timestamp: {}", timestamp);	1✔
157	stopReadToCache();	1✔
158	try {
159	Optional<Long> positionOptional = findPosition(timestamp);	1✔
160	if (!positionOptional.isPresent()) {	1✔
161	return false;	1✔
162	}
163	Long position = positionOptional.get();	1✔
164	return moveToPositionInternal(position);	1✔
165	} finally {
166	startReadToCache();	1✔
167	logDebug("[moveToTimestamp] end");	1✔
168	}	×
169	}
170
171	@Override
172	public Optional<QueueMessage> get(final long position) {
173	if (position < 0) {	1✔
NEW 174	return Optional.empty();	×
175	}
176	try (ExcerptTailer tailer = queue.createTailer()) {	1✔
177	tailer.moveToIndex(position);	1✔
178	InternalReadMessage internalReadMessage = new InternalReadMessage();	1✔
179	boolean readResult = tailer.readBytes(internalReadMessage);	1✔
180	if (readResult) {	1✔
181	return Optional.of(toQueueMessage(internalReadMessage, position));	1✔
182	} else {
NEW 183	return Optional.empty();	×
184	}
185	}	1✔
186	}
187
188	@Override
189	public Optional<QueueMessage> get(final String messageKey, long searchTimestampStart, long searchTimestampEnd) {
190	if (messageKey == null \|\| messageKey.isEmpty()) {	1✔
191	return Optional.empty();	1✔
192	}
193	try (ExcerptTailer tailer = queue.createTailer()) {	1✔
194	moveToNearByTimestamp(tailer, searchTimestampStart);	1✔
195	while (true) {
196	// for performance, ignore read content.
197	InternalReadMessage internalReadMessage = new InternalReadMessage(true);	1✔
198	boolean readResult = tailer.readBytes(internalReadMessage);	1✔
199	if (!readResult) {	1✔
200	return Optional.empty();	1✔
201	}
202	if (internalReadMessage.getWriteTime() < searchTimestampStart) {	1✔
203	continue;	1✔
204	}
205	if (internalReadMessage.getWriteTime() > searchTimestampEnd) {	1✔
206	return Optional.empty();	×
207	}
208	boolean moveToResult = tailer.moveToIndex(tailer.lastReadIndex());	1✔
209	if (!moveToResult) {	1✔
210	return Optional.empty();	×
211	}
212	internalReadMessage = new InternalReadMessage();	1✔
213	readResult = tailer.readBytes(internalReadMessage);	1✔
214	if (!readResult) {	1✔
215	return Optional.empty();	×
216	}
217	QueueMessage queueMessage = toQueueMessage(internalReadMessage, tailer.lastReadIndex());	1✔
218	if (Objects.equals(messageKey, queueMessage.getMessageKey())) {	1✔
219	return Optional.of(queueMessage);	1✔
220	}
UNCOV 221	}	×
222	}	1✔
223	}
224
225	private QueueMessage toQueueMessage(final InternalReadMessage internalReadMessage, final long position) {
226	return new QueueMessage(	1✔
227	internalReadMessage.getMessageKey(),	1✔
228	positionVersion.get(),	1✔
229	position,
230	internalReadMessage.getContent(),	1✔
231	internalReadMessage.getWriteTime());	1✔
232	}
233
234	private boolean moveToPositionInternal(final long position) {
235	return CompletableFuture.supplyAsync(() -> {	1✔
236	synchronized (closeLocker) {	1✔
237	try {
238	if (isClosing.get()) {	1✔
NEW 239	logDebug("[moveToPositionInternal] consumer is closing");	×
NEW 240	return false;	×
241	}
242	logDebug("[moveToPositionInternal] start, position: {}", position);	1✔
243	boolean moveToResult = mainTailer.moveToIndex(position);	1✔
244	if (moveToResult) {	1✔
245	positionVersion.incrementAndGet();	1✔
246	messageCache.clear();	1✔
247	ackedReadPosition.set(position);	1✔
248	}
249	return moveToResult;	1✔
250	} finally {
251	logDebug("[moveToPositionInternal] end");	1✔
UNCOV 252	}	×
UNCOV 253	}	×
254	}, this.readCacheExecutor).join();	1✔
255	}
256
257
258	@Override
259	public Optional<Long> findPosition(final long timestamp) {
260	logDebug("[findPosition] start, timestamp: {}", timestamp);	1✔
261	try (ExcerptTailer tailer = queue.createTailer()) {	1✔
262	moveToNearByTimestamp(tailer, timestamp);	1✔
263	while (true) {
264	InternalReadMessage internalReadMessage = new InternalReadMessage(true);	1✔
265	boolean resultResult = tailer.readBytes(internalReadMessage);	1✔
266	if (resultResult) {	1✔
267	if (internalReadMessage.getWriteTime() >= timestamp) {	1✔
268	return Optional.of(tailer.lastReadIndex());	1✔
269	}
270	} else {
271	return Optional.empty();	1✔
272	}
273	}	1✔
274	} finally {	1✔
275	logDebug("[findPosition] end");	1✔
276	}	×
277	}
278
279	public long getAckedReadPosition() {
280	return ackedReadPosition.get();	1✔
281	}
282
283	@Override
284	public boolean isClosed() {
285	return isClosed.get();	1✔
286	}
287
288	private void stopReadToCache() {
289	isReadToCacheRunning.set(false);	1✔
290	}	1✔
291
292	private void startReadToCache() {
293	this.isReadToCacheRunning.set(true);	1✔
294	readCacheExecutor.execute(this::readToCache);	1✔
295	}	1✔
296
297	private void readToCache() {
298	try {
299	logDebug("[readToCache] start");	1✔
300	long pullInterval = config.getPullInterval();	1✔
301	long fillCacheInterval = config.getFillCacheInterval();	1✔
302	while (isReadToCacheRunning.get() && !isClosing.get()) {	1✔
303	synchronized (closeLocker) {	1✔
304	try {
305	if (isClosing.get()) {	1✔
306	logDebug("[readToCache] consumer is closing");	1✔
307	return;	1✔
308	}
309	InternalReadMessage internalReadMessage = new InternalReadMessage();	1✔
310	boolean readResult = mainTailer.readBytes(internalReadMessage);	1✔
311	if (!readResult) {	1✔
312	TimeUnit.MILLISECONDS.sleep(pullInterval);	1✔
313	continue;	1✔
314	}
315	long lastedReadIndex = mainTailer.lastReadIndex();	1✔
316	QueueMessage queueMessage = toQueueMessage(internalReadMessage, lastedReadIndex);	1✔
317	boolean offerResult = this.messageCache.offer(queueMessage, fillCacheInterval, TimeUnit.MILLISECONDS);	1✔
318	if (!offerResult) {	1✔
319	// if offer failed, move to last read position
320	mainTailer.moveToIndex(lastedReadIndex);	1✔
321	}
NEW 322	} catch (InterruptedException e) {	×
NEW 323	Thread.currentThread().interrupt();	×
324	}	1✔
325	}	1✔
326	}
327	} finally {
328	logDebug("[readToCache] end");	1✔
329	}	1✔
330	}	1✔
331
332	private ExcerptTailer initMainTailer() {
333	return CompletableFuture.supplyAsync(this::initMainTailerInternal, this.readCacheExecutor).join();	1✔
334	}
335
336	private ExcerptTailer initMainTailerInternal() {
337	try {
338	logDebug("[initExcerptTailerInternal] start");	1✔
339	ExcerptTailer tailer = queue.createTailer();	1✔
340	Optional<Long> lastPositionOptional = getLastPosition();	1✔
341	if (lastPositionOptional.isPresent()) {	1✔
342	Long position = lastPositionOptional.get();	1✔
343	long beginPosition = position + 1;	1✔
344	tailer.moveToIndex(beginPosition);	1✔
345	logDebug("[initExcerptTailerInternal] find last position and move to position: {}", beginPosition);	1✔
346	} else {	1✔
347	ConsumeFromWhere consumeFromWhere = this.config.getConsumeFromWhere();	1✔
348	if (consumeFromWhere == ConsumeFromWhere.LAST) {	1✔
349	tailer.toEnd();	1✔
350	logDebug("[initExcerptTailerInternal] move to end");	1✔
351	} else if (consumeFromWhere == ConsumeFromWhere.FIRST) {	1✔
352	tailer.toStart();	1✔
353	logDebug("[initExcerptTailerInternal] move to start");	1✔
354	}
355	}
356	return tailer;	1✔
357	} finally {
358	logDebug("[initExcerptTailer] end");	1✔
359	}	×
360
361	}
362
363	/// region position
364
365	private void flushPosition() {
366	if (ackedReadPosition.get() != -1) {	1✔
367	setLastPosition(this.ackedReadPosition.get());	1✔
368	}
369	}	1✔
370
371	private Optional<Long> getLastPosition() {
372	return positionStore.get(config.getConsumerId());	1✔
373	}
374
375	private void setLastPosition(long position) {
376	positionStore.put(config.getConsumerId(), position);	1✔
377	}	1✔
378
379	/// endregion
380
381	@SuppressWarnings("Duplicates")
382	@Override
383	public void close() {
384	synchronized (closeLocker) {	1✔
385	try {
386	logDebug("[close] start");	1✔
387	if (isClosing.get()) {	1✔
388	logDebug("[close] is closing");	1✔
389	return;	1✔
390	}
391	isClosing.set(true);	1✔
392	stopReadToCache();	1✔
393	if (!positionStore.isClosed()) {	1✔
394	positionStore.close();	1✔
395	}
396	scheduler.shutdown();	1✔
397	readCacheExecutor.shutdown();	1✔
398	try {
399	if (!scheduler.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
NEW 400	scheduler.shutdownNow();	×
401	}
402	if (!readCacheExecutor.awaitTermination(1, TimeUnit.SECONDS)) {	1✔
403	readCacheExecutor.shutdownNow();	1✔
404	}
NEW 405	} catch (InterruptedException e) {	×
NEW 406	scheduler.shutdownNow();	×
UNCOV 407	readCacheExecutor.shutdownNow();	×
NEW 408	Thread.currentThread().interrupt();	×
409	}	1✔
410	if (!queue.isClosed()) {	1✔
411	queue.close();	1✔
412	}
413
414	for (CloseListener closeListener : closeListenerList) {	1✔
415	closeListener.onClose();	1✔
416	}	1✔
417	isClosed.set(true);	1✔
418	} finally {
419	logDebug("[close] end");	1✔
420	}	1✔
421	}	1✔
422	}	1✔
423
424	private void moveToNearByTimestamp(ExcerptTailer tailer, long timestamp) {
425	int expectedCycle = ChronicleQueueHelper.cycle(defaultRollCycle, timeProvider, timestamp);	1✔
426	int currentCycle = tailer.cycle();	1✔
427	if (currentCycle != expectedCycle) {	1✔
428	boolean moveToCycleResult = tailer.moveToCycle(expectedCycle);	1✔
429	logDebug("[moveToNearByTimestamp] moveToCycleResult: {}", moveToCycleResult);	1✔
430	}
431	}	1✔
432
433	@Override
434	public void addCloseListener(CloseListener listener) {
435	closeListenerList.add(listener);	1✔
436	}	1✔
437
438
439	// region page
440
441	@Override
442	public PageInfo<QueueMessage> getPage(SortDirection sortDirection, int pageSize) {
443	return getPage(-1, sortDirection, pageSize);	1✔
444	}
445
446	@SuppressWarnings("Duplicates")
447	@Override
448	public PageInfo<QueueMessage> getPage(long moveToPosition, SortDirection sortDirection, int pageSize) {
449	try (ExcerptTailer tailer = queue.createTailer()) {	1✔
450	if (moveToPosition != -1) {	1✔
451	tailer.moveToIndex(moveToPosition);	1✔
452	}
453	if (sortDirection == SortDirection.DESC) {	1✔
454	tailer.toEnd();	1✔
455	tailer.direction(TailerDirection.BACKWARD);	1✔
456	}
457	List<QueueMessage> data = new ArrayList<>();	1✔
458	long start = -1;	1✔
459	long end = -1;	1✔
460	for (int i = 0; i < pageSize; i++) {	1✔
461	InternalReadMessage internalReadMessage = new InternalReadMessage();	1✔
462	boolean readResult = tailer.readBytes(internalReadMessage);	1✔
463	if (!readResult) {	1✔
464	break;	1✔
465	}
466	QueueMessage queueMessage = toQueueMessage(internalReadMessage, tailer.lastReadIndex());	1✔
467	data.add(queueMessage);	1✔
468	if (i == 0) {	1✔
469	start = tailer.lastReadIndex();	1✔
470	}
471	end = tailer.lastReadIndex();	1✔
472	}
473	return new PageInfo<>(start, end, data, sortDirection, pageSize);	1✔
474	}	1✔
475	}
476
477	@SuppressWarnings("Duplicates")
478	@Override
479	public PageInfo<QueueMessage> getPage(PageInfo<QueueMessage> prevPageInfo, UpDown upDown) {
480	SortDirection sortDirection = prevPageInfo.getSortDirection();	1✔
481	int pageSize = prevPageInfo.getPageSize();	1✔
482	long start = prevPageInfo.getStart();	1✔
483	long end = prevPageInfo.getEnd();	1✔
484	try (ExcerptTailer tailer = queue.createTailer()) {	1✔
485	TailerDirection tailerDirection = getTailerDirection(sortDirection, upDown);	1✔
486	tailer.direction(tailerDirection);	1✔
487	if (sortDirection == SortDirection.DESC) {	1✔
488	if (upDown == UpDown.DOWN) {	1✔
489	tailer.moveToIndex(end - 1);	1✔
490	} else {
491	tailer.moveToIndex(start + 1);	1✔
492	}
493	} else {
494	if (upDown == UpDown.DOWN) {	1✔
495	tailer.moveToIndex(end + 1);	1✔
496	} else {
497	tailer.moveToIndex(start - 1);	1✔
498	}
499	}
500	List<QueueMessage> data = new ArrayList<>();	1✔
501	for (int i = 0; i < pageSize; i++) {	1✔
502	InternalReadMessage internalReadMessage = new InternalReadMessage();	1✔
503	boolean readResult = tailer.readBytes(internalReadMessage);	1✔
504	if (!readResult) {	1✔
NEW 505	break;	×
506	}
507	QueueMessage queueMessage = toQueueMessage(internalReadMessage, tailer.lastReadIndex());	1✔
508	data.add(queueMessage);	1✔
509	if (i == 0) {	1✔
510	start = tailer.lastReadIndex();	1✔
511	}
512	end = tailer.lastReadIndex();	1✔
513	}
514	if (upDown == UpDown.UP) {	1✔
515	Collections.reverse(data);	1✔
516	}
517	return new PageInfo<>(start, end, data, sortDirection, pageSize);	1✔
518	}	1✔
519	}
520
521	private TailerDirection getTailerDirection(SortDirection sortDirection, UpDown upDown) {
522	if (sortDirection == SortDirection.DESC && upDown == UpDown.DOWN) {	1✔
523	return TailerDirection.BACKWARD;	1✔
524	}
525	if (sortDirection == SortDirection.DESC && upDown == UpDown.UP) {	1✔
526	return TailerDirection.FORWARD;	1✔
527	}
528	if (sortDirection == SortDirection.ASC && upDown == UpDown.DOWN) {	1✔
529	return TailerDirection.FORWARD;	1✔
530	}
531	if (sortDirection == SortDirection.ASC && upDown == UpDown.UP) {	1✔
532	return TailerDirection.BACKWARD;	1✔
533	}
NEW 534	return TailerDirection.FORWARD;	×
535	}
536
537
538	// endregion
539
540	// region logger
541
542	private void logDebug(String format) {
543	if (logger.isDebugEnabled()) {	1✔
544	logger.debug(format);	×
545	}
546	}	1✔
547
548	private void logDebug(String format, Object arg) {
549	if (logger.isDebugEnabled()) {	1✔
550	logger.debug(format, arg);	×
551	}
552	}	1✔
553
554	// endregion
555	}

wz2cool / local-queue / #48

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