wz2cool / local-queue / #54

Committed 03 Feb 2025 03:31PM UTC coverage: 92.219% (+0.3%) from 91.935%

Build # #54

Build Type

push

Committed by

Commit Message

Merge pull request #8 from wz2cool/0.2.0

0.2.0

Coverage Stats

49 of 50 new or added lines in 8 files covered. (98.0%)

723 of 784 relevant lines covered (92.22%)

0.92 hits per line

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

91.05

/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 Set<String> matchTags;
    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.matchTags = getMatchTags(config.getSelectTag());
        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 Set<String> getMatchTags(String selectTag) {
        logDebug("[getMatchTags] start, selectTag: {}", selectTag);
        ConcurrentHashMap.KeySetView<String, Boolean> mySet = ConcurrentHashMap.newKeySet();
        if (selectTag == null || selectTag.isEmpty()) {
            return mySet;
        }

        String[] tags = selectTag.split("\\|\\|");
        mySet.addAll(Arrays.asList(tags));
        return mySet;
    }

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

wz2cool / local-queue / #54

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