#431

Committed 28 Apr 2025 05:27AM UTC coverage: 88.727% (-0.08%) from 88.802%

Build # #431

Build Type

push

Committed by

Commit Message

ci: change version to 2.7.9-SNAPSHOT

Run Details

4754 of 5358 relevant lines covered (88.73%)

0.89 hits per line

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

67.39

/jetcache-core/src/main/java/com/alicp/jetcache/support/BroadcastManager.java

/**
 * Created on 2019/6/10.
 */
package com.alicp.jetcache.support;

import com.alicp.jetcache.Cache;
import com.alicp.jetcache.CacheManager;
import com.alicp.jetcache.CacheResult;
import com.alicp.jetcache.CacheUtil;
import com.alicp.jetcache.MultiLevelCache;
import com.alicp.jetcache.embedded.AbstractEmbeddedCache;
import com.alicp.jetcache.external.ExternalCacheConfig;
import com.alicp.jetcache.CacheConfigException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.UUID;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author huangli
 */
public abstract class BroadcastManager implements AutoCloseable {
    private static Logger logger = LoggerFactory.getLogger(BroadcastManager.class);

    private final String sourceId = UUID.randomUUID().toString();
    private final CacheManager cacheManager;

    public BroadcastManager(CacheManager cacheManager) {
        this.cacheManager = cacheManager;
    }

    protected void checkConfig(ExternalCacheConfig config) {
        if (config.getBroadcastChannel() == null) {
            throw new CacheConfigException("BroadcastChannel not set");
        }
        if (config.getValueEncoder() == null) {
            throw new CacheConfigException("no value encoder");
        }
        if (config.getValueDecoder() == null) {
            throw new CacheConfigException("no value decoder");
        }
    }

    public abstract CacheResult publish(CacheMessage cacheMessage);

    public abstract void startSubscribe();

    @Override
    public void close() throws Exception {
    }

    public String getSourceId() {
        return sourceId;
    }

    public CacheManager getCacheManager() {
        return cacheManager;
    }

    protected void processNotification(byte[] message, Function<byte[], Object> decoder) {
        try {
            if (message == null) {
                logger.error("notify message is null");
                return;
            }
            Object value = decoder.apply(message);
            if (value == null) {
                logger.error("notify message is null");
                return;
            }
            if (value instanceof CacheMessage) {
                processCacheMessage((CacheMessage) value);
            } else {
                logger.error("the message is not instance of CacheMessage, class={}", value.getClass());
            }
        } catch (Throwable e) {
            SquashedLogger.getLogger(logger).error("receive cache notify error", e);
        }
    }

    private void processCacheMessage(CacheMessage cacheMessage) {
        if (sourceId.equals(cacheMessage.getSourceId())) {
            return;
        }
        Cache cache = cacheManager.getCache(cacheMessage.getArea(), cacheMessage.getCacheName());
        if (cache == null) {
            logger.warn("Cache instance not exists: {},{}", cacheMessage.getArea(), cacheMessage.getCacheName());
            return;
        }
        Cache absCache = CacheUtil.getAbstractCache(cache);
        if (!(absCache instanceof MultiLevelCache)) {
            logger.warn("Cache instance is not MultiLevelCache: {},{}", cacheMessage.getArea(), cacheMessage.getCacheName());
            return;
        }
        Cache[] caches = ((MultiLevelCache) absCache).caches();
        Set<Object> keys = Stream.of(cacheMessage.getKeys()).collect(Collectors.toSet());
        for (Cache c : caches) {
            Cache localCache = CacheUtil.getAbstractCache(c);
            if (localCache instanceof AbstractEmbeddedCache) {
                ((AbstractEmbeddedCache) localCache).__removeAll(keys);
            } else {
                break;
            }
        }
    }

}

1	/**
2	* Created on 2019/6/10.
3	*/
4	package com.alicp.jetcache.support;
5
6	import com.alicp.jetcache.Cache;
7	import com.alicp.jetcache.CacheManager;
8	import com.alicp.jetcache.CacheResult;
9	import com.alicp.jetcache.CacheUtil;
10	import com.alicp.jetcache.MultiLevelCache;
11	import com.alicp.jetcache.embedded.AbstractEmbeddedCache;
12	import com.alicp.jetcache.external.ExternalCacheConfig;
13	import com.alicp.jetcache.CacheConfigException;
14	import org.slf4j.Logger;
15	import org.slf4j.LoggerFactory;
16
17	import java.util.Set;
18	import java.util.UUID;
19	import java.util.function.Function;
20	import java.util.stream.Collectors;
21	import java.util.stream.Stream;
22
23	/**
24	* @author huangli
25	*/
26	public abstract class BroadcastManager implements AutoCloseable {
27	private static Logger logger = LoggerFactory.getLogger(BroadcastManager.class);	1✔
28
29	private final String sourceId = UUID.randomUUID().toString();	1✔
30	private final CacheManager cacheManager;
31
32	public BroadcastManager(CacheManager cacheManager) {	1✔
33	this.cacheManager = cacheManager;	1✔
34	}	1✔
35
36	protected void checkConfig(ExternalCacheConfig config) {
37	if (config.getBroadcastChannel() == null) {	1✔
38	throw new CacheConfigException("BroadcastChannel not set");	×
39	}
40	if (config.getValueEncoder() == null) {	1✔
41	throw new CacheConfigException("no value encoder");	×
42	}
43	if (config.getValueDecoder() == null) {	1✔
44	throw new CacheConfigException("no value decoder");	×
45	}
46	}	1✔
47
48	public abstract CacheResult publish(CacheMessage cacheMessage);
49
50	public abstract void startSubscribe();
51
52	@Override
53	public void close() throws Exception {
54	}	×
55
56	public String getSourceId() {
57	return sourceId;	1✔
58	}
59
60	public CacheManager getCacheManager() {
61	return cacheManager;	1✔
62	}
63
64	protected void processNotification(byte[] message, Function<byte[], Object> decoder) {
65	try {
66	if (message == null) {	1✔
67	logger.error("notify message is null");	×
68	return;	×
69	}
70	Object value = decoder.apply(message);	1✔
71	if (value == null) {	1✔
72	logger.error("notify message is null");	×
73	return;	×
74	}
75	if (value instanceof CacheMessage) {	1✔
76	processCacheMessage((CacheMessage) value);	1✔
77	} else {
78	logger.error("the message is not instance of CacheMessage, class={}", value.getClass());	×
79	}
80	} catch (Throwable e) {	×
81	SquashedLogger.getLogger(logger).error("receive cache notify error", e);	×
82	}	1✔
83	}	1✔
84
85	private void processCacheMessage(CacheMessage cacheMessage) {
86	if (sourceId.equals(cacheMessage.getSourceId())) {	1✔
87	return;	1✔
88	}
89	Cache cache = cacheManager.getCache(cacheMessage.getArea(), cacheMessage.getCacheName());	1✔
90	if (cache == null) {	1✔
91	logger.warn("Cache instance not exists: {},{}", cacheMessage.getArea(), cacheMessage.getCacheName());	×
92	return;	×
93	}
94	Cache absCache = CacheUtil.getAbstractCache(cache);	1✔
95	if (!(absCache instanceof MultiLevelCache)) {	1✔
96	logger.warn("Cache instance is not MultiLevelCache: {},{}", cacheMessage.getArea(), cacheMessage.getCacheName());	×
97	return;	×
98	}
99	Cache[] caches = ((MultiLevelCache) absCache).caches();	1✔
100	Set<Object> keys = Stream.of(cacheMessage.getKeys()).collect(Collectors.toSet());	1✔
101	for (Cache c : caches) {	1✔
102	Cache localCache = CacheUtil.getAbstractCache(c);	1✔
103	if (localCache instanceof AbstractEmbeddedCache) {	1✔
104	((AbstractEmbeddedCache) localCache).__removeAll(keys);	1✔
105	} else {
106	break;
107	}
108	}
109	}	1✔
110
111	}

alibaba / jetcache / #431

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