wz2cool / local-queue / #28

Committed 31 Jan 2025 11:37AM UTC coverage: 91.871% (+0.4%) from 91.466%

Build # #28

Build Type

push

Committed by

Commit Message

Merge pull request #5 from wz2cool/0.1.1

0.1.1

Coverage Stats

92 of 96 new or added lines in 8 files covered. (95.83%)

486 of 529 relevant lines covered (91.87%)

0.92 hits per line

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

94.59

/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 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 SimpleQueueConfig config;
    private final SimpleProducer simpleProducer;
    private final Map<String, SimpleConsumer> consumerMap = new ConcurrentHashMap<>();
    private final ConcurrentLinkedQueue<CloseListener> closeListeners = new ConcurrentLinkedQueue<>();

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

    @Override
    public void close() {
        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();
        }
    }

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

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
11	import java.util.Map;
12	import java.util.Objects;
13	import java.util.concurrent.ConcurrentHashMap;
14	import java.util.concurrent.ConcurrentLinkedQueue;
15
16	/**
17	* simple queue
18	*
19	* @author frank
20	*/
21	public class SimpleQueue implements IQueue {
22
23	private final SimpleQueueConfig config;
24	private final SimpleProducer simpleProducer;
25	private final Map<String, SimpleConsumer> consumerMap = new ConcurrentHashMap<>();	1✔
26	private final ConcurrentLinkedQueue<CloseListener> closeListeners = new ConcurrentLinkedQueue<>();	1✔
27
28	public SimpleQueue(SimpleQueueConfig config) {	1✔
29	this.config = config;	1✔
30	this.simpleProducer = getProducer();	1✔
31	}	1✔
32
33	@Override
34	public boolean offer(String message) {
35	return simpleProducer.offer(message);	1✔
36	}
37
38	@Override
39	public boolean offer(String messageKey, String message) {
NEW 40	return simpleProducer.offer(messageKey, message);	×
41	}
42
43	private SimpleProducer getProducer() {
44	return new SimpleProducer(new SimpleProducerConfig.Builder()	1✔
45	.setDataDir(config.getDataDir())	1✔
46	.setKeepDays(config.getKeepDays())	1✔
47	.build());	1✔
48	}
49
50	@Override
51	public synchronized IConsumer getConsumer(final String consumerId) {
52	return getConsumer(consumerId, ConsumeFromWhere.LAST);	1✔
53	}
54
55	@Override
56	public synchronized IConsumer getConsumer(final String consumerId, final ConsumeFromWhere consumeFromWhere) {
57	SimpleConsumer consumer = consumerMap.get(consumerId);	1✔
58	if (Objects.nonNull(consumer)) {	1✔
59	return consumer;	×
60	}
61
62	consumer = new SimpleConsumer(new SimpleConsumerConfig.Builder()	1✔
63	.setDataDir(config.getDataDir())	1✔
64	.setConsumerId(consumerId)	1✔
65	.setConsumeFromWhere(consumeFromWhere)	1✔
66	.build());	1✔
67	consumer.addCloseListener(() -> consumerMap.remove(consumerId));	1✔
68	consumerMap.put(consumerId, consumer);	1✔
69	return consumer;	1✔
70	}
71
72	@Override
73	public void close() {
74	if (!simpleProducer.isClosed()) {	1✔
75	simpleProducer.close();	1✔
76	}
77	for (Map.Entry<String, SimpleConsumer> entry : consumerMap.entrySet()) {	1✔
78	SimpleConsumer consumer = entry.getValue();	1✔
79	if (!consumer.isClosed()) {	1✔
80	entry.getValue().close();	1✔
81	}
82	}	1✔
83	for (CloseListener listener : closeListeners) {	1✔
84	listener.onClose();	1✔
85	}	1✔
86	}	1✔
87
88	@Override
89	public void addCloseListener(CloseListener listener) {
90	closeListeners.add(listener);	1✔
91	}	1✔
92	}

wz2cool / local-queue / #28

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