wz2cool / local-queue / #38

Committed 01 Feb 2025 12:36PM UTC coverage: 91.85% (-0.02%) from 91.871%

Build # #38

Build Type

push

Committed by

Commit Message

Merge pull request #6 from wz2cool/0.1.2

0.1.2

Coverage Stats

113 of 127 new or added lines in 9 files covered. (88.98%)

586 of 638 relevant lines covered (91.85%)

0.92 hits per line

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

94.34

/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;

/**
 * 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 volatile boolean isClosed = 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);
    }

    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(() -> consumerMap.remove(consumerId));
        consumerMap.put(consumerId, consumer);
        return consumer;
    }

    @Override
    public void close() {
        try {
            logDebug("[close] start");
            if (isClosed) {
                logDebug("[close] already closed");
                return;
            }
            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 = 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
18	/**
19	* simple queue
20	*
21	* @author frank
22	*/
23	public class SimpleQueue implements IQueue {
24
25	private final Logger logger = LoggerFactory.getLogger(this.getClass());	1✔
26	private final SimpleQueueConfig config;
27	private final SimpleProducer simpleProducer;
28	private final Map<String, SimpleConsumer> consumerMap = new ConcurrentHashMap<>();	1✔
29	private final ConcurrentLinkedQueue<CloseListener> closeListeners = new ConcurrentLinkedQueue<>();	1✔
30	private volatile boolean isClosed = false;	1✔
31
32	public SimpleQueue(SimpleQueueConfig config) {	1✔
33	this.config = config;	1✔
34	this.simpleProducer = getProducer();	1✔
35	}	1✔
36
37	@Override
38	public boolean offer(String message) {
39	return simpleProducer.offer(message);	1✔
40	}
41
42	@Override
43	public boolean offer(String messageKey, String message) {
44	return simpleProducer.offer(messageKey, message);	×
45	}
46
47	private SimpleProducer getProducer() {
48	return new SimpleProducer(new SimpleProducerConfig.Builder()	1✔
49	.setDataDir(config.getDataDir())	1✔
50	.setKeepDays(config.getKeepDays())	1✔
51	.setRollCycleType(config.getRollCycleType())	1✔
52	.setTimeZone(config.getTimeZone())	1✔
53	.build());	1✔
54	}
55
56	@Override
57	public synchronized IConsumer getConsumer(final String consumerId) {
58	return getConsumer(consumerId, ConsumeFromWhere.LAST);	1✔
59	}
60
61	@Override
62	public synchronized IConsumer getConsumer(final String consumerId, final ConsumeFromWhere consumeFromWhere) {
63	SimpleConsumer consumer = consumerMap.get(consumerId);	1✔
64	if (Objects.nonNull(consumer)) {	1✔
65	return consumer;	×
66	}
67
68	consumer = new SimpleConsumer(new SimpleConsumerConfig.Builder()	1✔
69	.setDataDir(config.getDataDir())	1✔
70	.setConsumerId(consumerId)	1✔
71	.setConsumeFromWhere(consumeFromWhere)	1✔
72	.setRollCycleType(config.getRollCycleType())	1✔
73	.setTimeZone(config.getTimeZone())	1✔
74	.build());	1✔
75	consumer.addCloseListener(() -> consumerMap.remove(consumerId));	1✔
76	consumerMap.put(consumerId, consumer);	1✔
77	return consumer;	1✔
78	}
79
80	@Override
81	public void close() {
82	try {
83	logDebug("[close] start");	1✔
84	if (isClosed) {	1✔
85	logDebug("[close] already closed");	1✔
86	return;	1✔
87	}
88	if (!simpleProducer.isClosed()) {	1✔
89	simpleProducer.close();	1✔
90	}
91	for (Map.Entry<String, SimpleConsumer> entry : consumerMap.entrySet()) {	1✔
92	SimpleConsumer consumer = entry.getValue();	1✔
93	if (!consumer.isClosed()) {	1✔
94	entry.getValue().close();	1✔
95	}
96	}	1✔
97	for (CloseListener listener : closeListeners) {	1✔
98	listener.onClose();	1✔
99	}	1✔
100	isClosed = true;	1✔
101	} finally {
102	logDebug("[close] end");	1✔
103	}	1✔
104	}	1✔
105
106
107	@Override
108	public void addCloseListener(CloseListener listener) {
109	closeListeners.add(listener);	1✔
110	}	1✔
111
112	// region logger
113
114	private void logDebug(String format) {
115	if (logger.isDebugEnabled()) {	1✔
NEW 116	logger.debug(format);	×
117	}
118	}	1✔
119
120	// endregion
121	}

wz2cool / local-queue / #38

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