#421

Committed 12 Sep 2024 08:19AM UTC coverage: 88.785% (-0.06%) from 88.841%

Build # #421

Build Type

push

Committed by

Commit Message

ci: change pom version to 2.7.8-SNAPSHOT

Run Details

4750 of 5350 relevant lines covered (88.79%)

0.89 hits per line

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

77.61

/jetcache-support/jetcache-redis/src/main/java/com/alicp/jetcache/redis/RedisBroadcastManager.java

package com.alicp.jetcache.redis;

import com.alicp.jetcache.CacheConfigException;
import com.alicp.jetcache.CacheManager;
import com.alicp.jetcache.CacheResult;
import com.alicp.jetcache.support.BroadcastManager;
import com.alicp.jetcache.support.CacheMessage;
import com.alicp.jetcache.support.SquashedLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.BinaryJedisPubSub;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.UnifiedJedis;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created on 2022-05-03
 *
 * @author huangli
 */
public class RedisBroadcastManager extends BroadcastManager {

    private static final Logger logger = LoggerFactory.getLogger(RedisBroadcastManager.class);

    private final byte[] channel;
    private final String channelStr;
    private final RedisCacheConfig<Object, Object> config;

    private volatile CacheMessagePubSub cacheMessagePubSub;
    private volatile boolean closed;
    private volatile boolean subscribe;
    private boolean subscribeThreadStart;

    private final ReentrantLock reentrantLock = new ReentrantLock();

    public RedisBroadcastManager(CacheManager cacheManager, RedisCacheConfig<Object, Object> config) {
        super(cacheManager);
        this.channelStr = config.getBroadcastChannel();
        this.channel = channelStr.getBytes(StandardCharsets.UTF_8);
        this.config = config;

        checkConfig(config);
        if (config.getJedis() == null && config.getJedisPool() == null) {
            throw new CacheConfigException("no jedis");
        }
        if (config.getJedis() != null && config.getJedisPool() != null) {
            throw new CacheConfigException("'jedis' and 'jedisPool' can't set simultaneously");
        }
    }

    @Override
    public void startSubscribe() {
        reentrantLock.lock();
        try {
            if (subscribeThreadStart) {
                throw new IllegalStateException("subscribe thread is started");
            }
            this.cacheMessagePubSub = new CacheMessagePubSub();
            Thread subThread;
            subThread = new Thread(this::runSubThread, "Sub_" + channelStr);
            subThread.setDaemon(true);
            subThread.start();
            this.subscribeThreadStart = true;
        }finally {
            reentrantLock.unlock();
        }
    }

    private void runSubThread() {
        while (!closed) {
            runSubThread0();
        }
    }

    private void runSubThread0() {
        Object jedisObj = null;
        try {
            jedisObj = writeCommands();
            if (jedisObj instanceof Jedis) {
                subscribe = true;
                ((Jedis) jedisObj).subscribe(cacheMessagePubSub, channel);
            } else if (jedisObj instanceof UnifiedJedis) {
                subscribe = true;
                ((UnifiedJedis) jedisObj).subscribe(cacheMessagePubSub, channel);
            }
        } catch (Throwable e) {
            SquashedLogger.getLogger(logger).error("run jedis subscribe thread error: {}", e);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                // ignore
            }
        } finally {
            subscribe = false;
            RedisCache.closeJedis(jedisObj);
        }
    }

    Object writeCommands() {
        return config.getJedis() != null ? config.getJedis() : config.getJedisPool().getResource();
    }

    @Override
    public CacheResult publish(CacheMessage message) {
        Object jedisObj = null;
        try {
            jedisObj = writeCommands();
            byte[] value = config.getValueEncoder().apply(message);
            if (jedisObj instanceof Jedis) {
                ((Jedis) jedisObj).publish(channel, value);
            } else {
                ((UnifiedJedis) jedisObj).publish(channel, value);
            }
            return CacheResult.SUCCESS_WITHOUT_MSG;
        } catch (Exception ex) {
            SquashedLogger.getLogger(logger).error("jetcache publish error", ex);
            return new CacheResult(ex);
        } finally {
            RedisCache.closeJedis(jedisObj);
        }
    }


    @Override
    public void close() {
        reentrantLock.lock();
        try {
            if (this.closed) {
                return;
            }
            this.closed = true;
            if (subscribe) {
                try {
                    this.cacheMessagePubSub.unsubscribe(channel);
                } catch (Exception e) {
                    logger.warn("unsubscribe {} fail", channelStr, e);
                }
            }
        }finally {
            reentrantLock.unlock();
        }
    }

    class CacheMessagePubSub extends BinaryJedisPubSub {

        @Override
        public void onMessage(byte[] channel, byte[] message) {
            processNotification(message, config.getValueDecoder());
        }
    }
}

1	package com.alicp.jetcache.redis;
2
3	import com.alicp.jetcache.CacheConfigException;
4	import com.alicp.jetcache.CacheManager;
5	import com.alicp.jetcache.CacheResult;
6	import com.alicp.jetcache.support.BroadcastManager;
7	import com.alicp.jetcache.support.CacheMessage;
8	import com.alicp.jetcache.support.SquashedLogger;
9	import org.slf4j.Logger;
10	import org.slf4j.LoggerFactory;
11	import redis.clients.jedis.BinaryJedisPubSub;
12	import redis.clients.jedis.Jedis;
13	import redis.clients.jedis.UnifiedJedis;
14
15	import java.nio.charset.StandardCharsets;
16	import java.util.concurrent.locks.ReentrantLock;
17
18	/**
19	* Created on 2022-05-03
20	*
21	* @author huangli
22	*/
23	public class RedisBroadcastManager extends BroadcastManager {
24
25	private static final Logger logger = LoggerFactory.getLogger(RedisBroadcastManager.class);	1✔
26
27	private final byte[] channel;
28	private final String channelStr;
29	private final RedisCacheConfig<Object, Object> config;
30
31	private volatile CacheMessagePubSub cacheMessagePubSub;
32	private volatile boolean closed;
33	private volatile boolean subscribe;
34	private boolean subscribeThreadStart;
35
36	private final ReentrantLock reentrantLock = new ReentrantLock();	1✔
37
38	public RedisBroadcastManager(CacheManager cacheManager, RedisCacheConfig<Object, Object> config) {
39	super(cacheManager);	1✔
40	this.channelStr = config.getBroadcastChannel();	1✔
41	this.channel = channelStr.getBytes(StandardCharsets.UTF_8);	1✔
42	this.config = config;	1✔
43
44	checkConfig(config);	1✔
45	if (config.getJedis() == null && config.getJedisPool() == null) {	1✔
46	throw new CacheConfigException("no jedis");	×
47	}
48	if (config.getJedis() != null && config.getJedisPool() != null) {	1✔
49	throw new CacheConfigException("'jedis' and 'jedisPool' can't set simultaneously");	×
50	}
51	}	1✔
52
53	@Override
54	public void startSubscribe() {
55	reentrantLock.lock();	1✔
56	try {
57	if (subscribeThreadStart) {	1✔
58	throw new IllegalStateException("subscribe thread is started");	×
59	}
60	this.cacheMessagePubSub = new CacheMessagePubSub();	1✔
61	Thread subThread;
62	subThread = new Thread(this::runSubThread, "Sub_" + channelStr);	1✔
63	subThread.setDaemon(true);	1✔
64	subThread.start();	1✔
65	this.subscribeThreadStart = true;	1✔
66	}finally {
67	reentrantLock.unlock();	1✔
68	}
69	}	1✔
70
71	private void runSubThread() {
72	while (!closed) {	1✔
73	runSubThread0();	1✔
74	}
75	}	1✔
76
77	private void runSubThread0() {
78	Object jedisObj = null;	1✔
79	try {
80	jedisObj = writeCommands();	1✔
81	if (jedisObj instanceof Jedis) {	1✔
82	subscribe = true;	1✔
83	((Jedis) jedisObj).subscribe(cacheMessagePubSub, channel);	×
84	} else if (jedisObj instanceof UnifiedJedis) {	1✔
85	subscribe = true;	1✔
86	((UnifiedJedis) jedisObj).subscribe(cacheMessagePubSub, channel);	1✔
87	}
88	} catch (Throwable e) {	×
89	SquashedLogger.getLogger(logger).error("run jedis subscribe thread error: {}", e);	×
90	try {
91	Thread.sleep(1000);	×
92	} catch (InterruptedException ex) {	×
93	// ignore
94	}	×
95	} finally {
96	subscribe = false;	1✔
97	RedisCache.closeJedis(jedisObj);	1✔
98	}
99	}	1✔
100
101	Object writeCommands() {
102	return config.getJedis() != null ? config.getJedis() : config.getJedisPool().getResource();	1✔
103	}
104
105	@Override
106	public CacheResult publish(CacheMessage message) {
107	Object jedisObj = null;	1✔
108	try {
109	jedisObj = writeCommands();	1✔
110	byte[] value = config.getValueEncoder().apply(message);	1✔
111	if (jedisObj instanceof Jedis) {	1✔
112	((Jedis) jedisObj).publish(channel, value);	1✔
113	} else {
114	((UnifiedJedis) jedisObj).publish(channel, value);	1✔
115	}
116	return CacheResult.SUCCESS_WITHOUT_MSG;	1✔
117	} catch (Exception ex) {	×
118	SquashedLogger.getLogger(logger).error("jetcache publish error", ex);	×
119	return new CacheResult(ex);	×
120	} finally {
121	RedisCache.closeJedis(jedisObj);	1✔
122	}
123	}
124
125
126	@Override
127	public void close() {
128	reentrantLock.lock();	1✔
129	try {
130	if (this.closed) {	1✔
131	return;	×
132	}
133	this.closed = true;	1✔
134	if (subscribe) {	1✔
135	try {
136	this.cacheMessagePubSub.unsubscribe(channel);	1✔
137	} catch (Exception e) {	×
138	logger.warn("unsubscribe {} fail", channelStr, e);	×
139	}	1✔
140	}
141	}finally {
142	reentrantLock.unlock();	1✔
143	}
144	}	1✔
145
146	class CacheMessagePubSub extends BinaryJedisPubSub {	1✔
147
148	@Override
149	public void onMessage(byte[] channel, byte[] message) {
150	processNotification(message, config.getValueDecoder());	1✔
151	}	1✔
152	}
153	}

alibaba / jetcache / #421

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