wz2cool / local-queue / #41

Committed 02 Feb 2025 03:13PM UTC coverage: 91.958% (+0.2%) from 91.757%

Build # #41

Build Type

push

Committed by

Commit Message

use atomic instead of volatile

Run Details

105 of 124 new or added lines in 4 files covered. (84.68%)

2 existing lines in 2 files now uncovered.

606 of 659 relevant lines covered (91.96%)

0.92 hits per line

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

93.55

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

package com.github.wz2cool.localqueue.impl;

import com.github.wz2cool.localqueue.IConsumer;
import com.github.wz2cool.localqueue.IQueue;
import com.github.wz2cool.localqueue.event.CloseListener;
import com.github.wz2cool.localqueue.model.config.SimpleConsumerConfig;
import com.github.wz2cool.localqueue.model.config.SimpleProducerConfig;
import com.github.wz2cool.localqueue.model.config.SimpleQueueConfig;
import com.github.wz2cool.localqueue.model.enums.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * simple queue
 *
 * @author frank
 */
public class SimpleQueue implements IQueue {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private final SimpleQueueConfig config;
    private final SimpleProducer simpleProducer;
    private final Map<String, SimpleConsumer> consumerMap = new ConcurrentHashMap<>();
    private final ConcurrentLinkedQueue<CloseListener> closeListeners = new ConcurrentLinkedQueue<>();

    private final Object closeLocker = new Object();
    private final AtomicBoolean isClosing = new AtomicBoolean(false);
    private final AtomicBoolean isClosed = new AtomicBoolean(false);

    public SimpleQueue(SimpleQueueConfig config) {
        this.config = config;
        this.simpleProducer = getProducer();
    }

    @Override
    public boolean offer(String message) {
        return simpleProducer.offer(message);
    }

    @Override
    public boolean offer(String messageKey, String message) {
        return simpleProducer.offer(messageKey, message);
    }

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

    private SimpleProducer getProducer() {
        return new SimpleProducer(new SimpleProducerConfig.Builder()
                .setDataDir(config.getDataDir())
                .setKeepDays(config.getKeepDays())
                .setRollCycleType(config.getRollCycleType())
                .setTimeZone(config.getTimeZone())
                .build());
    }

    @Override
    public synchronized IConsumer getConsumer(final String consumerId) {
        return getConsumer(consumerId, ConsumeFromWhere.LAST);
    }

    @Override
    public synchronized IConsumer getConsumer(final String consumerId, final ConsumeFromWhere consumeFromWhere) {
        SimpleConsumer consumer = consumerMap.get(consumerId);
        if (Objects.nonNull(consumer)) {
            return consumer;
        }

        consumer = new SimpleConsumer(new SimpleConsumerConfig.Builder()
                .setDataDir(config.getDataDir())
                .setConsumerId(consumerId)
                .setConsumeFromWhere(consumeFromWhere)
                .setRollCycleType(config.getRollCycleType())
                .setTimeZone(config.getTimeZone())
                .build());
        consumer.addCloseListener(() -> {
            SimpleConsumer removeItem = consumerMap.remove(consumerId);
            removeItem.close();
        });
        consumerMap.put(consumerId, consumer);
        return consumer;
    }

    @Override
    public void close() {
        synchronized (closeLocker) {
            try {
                logDebug("[close] start");
                if (isClosing.get()) {
                    logDebug("[close] is closing");
                    return;
                }
                isClosing.set(true);
                if (!simpleProducer.isClosed()) {
                    simpleProducer.close();
                }
                for (Map.Entry<String, SimpleConsumer> entry : consumerMap.entrySet()) {
                    SimpleConsumer consumer = entry.getValue();
                    if (!consumer.isClosed()) {
                        entry.getValue().close();
                    }
                }
                for (CloseListener listener : closeListeners) {
                    listener.onClose();
                }
                isClosed.set(true);
            } finally {
                logDebug("[close] end");
            }
        }
    }


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

    // region logger

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

    // endregion
}

1	package com.github.wz2cool.localqueue.impl;
2
3	import com.github.wz2cool.localqueue.IConsumer;
4	import com.github.wz2cool.localqueue.IQueue;
5	import com.github.wz2cool.localqueue.event.CloseListener;
6	import com.github.wz2cool.localqueue.model.config.SimpleConsumerConfig;
7	import com.github.wz2cool.localqueue.model.config.SimpleProducerConfig;
8	import com.github.wz2cool.localqueue.model.config.SimpleQueueConfig;
9	import com.github.wz2cool.localqueue.model.enums.ConsumeFromWhere;
10	import org.slf4j.Logger;
11	import org.slf4j.LoggerFactory;
12
13	import java.util.Map;
14	import java.util.Objects;
15	import java.util.concurrent.ConcurrentHashMap;
16	import java.util.concurrent.ConcurrentLinkedQueue;
17	import java.util.concurrent.atomic.AtomicBoolean;
18
19	/**
20	* simple queue
21	*
22	* @author frank
23	*/
24	public class SimpleQueue implements IQueue {
25
26	private final Logger logger = LoggerFactory.getLogger(this.getClass());	1✔
27	private final SimpleQueueConfig config;
28	private final SimpleProducer simpleProducer;
29	private final Map<String, SimpleConsumer> consumerMap = new ConcurrentHashMap<>();	1✔
30	private final ConcurrentLinkedQueue<CloseListener> closeListeners = new ConcurrentLinkedQueue<>();	1✔
31
32	private final Object closeLocker = new Object();	1✔
33	private final AtomicBoolean isClosing = new AtomicBoolean(false);	1✔
34	private final AtomicBoolean isClosed = new AtomicBoolean(false);	1✔
35
36	public SimpleQueue(SimpleQueueConfig config) {	1✔
37	this.config = config;	1✔
38	this.simpleProducer = getProducer();	1✔
39	}	1✔
40
41	@Override
42	public boolean offer(String message) {
43	return simpleProducer.offer(message);	1✔
44	}
45
46	@Override
47	public boolean offer(String messageKey, String message) {
48	return simpleProducer.offer(messageKey, message);	×
49	}
50
51	@Override
52	public boolean isClosed() {
NEW 53	return isClosed.get();	×
54	}
55
56	private SimpleProducer getProducer() {
57	return new SimpleProducer(new SimpleProducerConfig.Builder()	1✔
58	.setDataDir(config.getDataDir())	1✔
59	.setKeepDays(config.getKeepDays())	1✔
60	.setRollCycleType(config.getRollCycleType())	1✔
61	.setTimeZone(config.getTimeZone())	1✔
62	.build());	1✔
63	}
64
65	@Override
66	public synchronized IConsumer getConsumer(final String consumerId) {
67	return getConsumer(consumerId, ConsumeFromWhere.LAST);	1✔
68	}
69
70	@Override
71	public synchronized IConsumer getConsumer(final String consumerId, final ConsumeFromWhere consumeFromWhere) {
72	SimpleConsumer consumer = consumerMap.get(consumerId);	1✔
73	if (Objects.nonNull(consumer)) {	1✔
74	return consumer;	×
75	}
76
77	consumer = new SimpleConsumer(new SimpleConsumerConfig.Builder()	1✔
78	.setDataDir(config.getDataDir())	1✔
79	.setConsumerId(consumerId)	1✔
80	.setConsumeFromWhere(consumeFromWhere)	1✔
81	.setRollCycleType(config.getRollCycleType())	1✔
82	.setTimeZone(config.getTimeZone())	1✔
83	.build());	1✔
84	consumer.addCloseListener(() -> {	1✔
85	SimpleConsumer removeItem = consumerMap.remove(consumerId);	1✔
86	removeItem.close();	1✔
87	});	1✔
88	consumerMap.put(consumerId, consumer);	1✔
89	return consumer;	1✔
90	}
91
92	@Override
93	public void close() {
94	synchronized (closeLocker) {	1✔
95	try {
96	logDebug("[close] start");	1✔
97	if (isClosing.get()) {	1✔
98	logDebug("[close] is closing");	1✔
99	return;	1✔
100	}
101	isClosing.set(true);	1✔
102	if (!simpleProducer.isClosed()) {	1✔
103	simpleProducer.close();	1✔
104	}
105	for (Map.Entry<String, SimpleConsumer> entry : consumerMap.entrySet()) {	1✔
106	SimpleConsumer consumer = entry.getValue();	1✔
107	if (!consumer.isClosed()) {	1✔
108	entry.getValue().close();	1✔
109	}
110	}	1✔
111	for (CloseListener listener : closeListeners) {	1✔
112	listener.onClose();	1✔
113	}	1✔
114	isClosed.set(true);	1✔
115	} finally {
116	logDebug("[close] end");	1✔
117	}	1✔
118	}	1✔
119	}	1✔
120
121
122	@Override
123	public void addCloseListener(CloseListener listener) {
124	closeListeners.add(listener);	1✔
125	}	1✔
126
127	// region logger
128
129	private void logDebug(String format) {
130	if (logger.isDebugEnabled()) {	1✔
131	logger.debug(format);	×
132	}
133	}	1✔
134
135	// endregion
136	}

wz2cool / local-queue / #41

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