1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.mybatis.caches.ignite;
17
18 import java.io.ByteArrayInputStream;
19 import java.io.ByteArrayOutputStream;
20 import java.io.IOException;
21 import java.io.InputStream;
22 import java.io.ObjectInputStream;
23 import java.io.ObjectOutputStream;
24 import java.nio.file.Files;
25 import java.nio.file.Path;
26 import java.util.Properties;
27 import java.util.concurrent.locks.ReadWriteLock;
28
29 import org.apache.ibatis.cache.Cache;
30 import org.apache.ibatis.logging.Log;
31 import org.apache.ibatis.logging.LogFactory;
32 import org.apache.ignite.catalog.ColumnType;
33 import org.apache.ignite.catalog.definitions.ColumnDefinition;
34 import org.apache.ignite.catalog.definitions.TableDefinition;
35 import org.apache.ignite.client.IgniteClient;
36 import org.apache.ignite.sql.ResultSet;
37 import org.apache.ignite.sql.SqlRow;
38 import org.apache.ignite.table.KeyValueView;
39 import org.apache.ignite.table.Tuple;
40
41
42
43
44
45
46
47 public final class IgniteCacheAdapter implements Cache {
48
49
50 private static final Log log = LogFactory.getLog(IgniteCacheAdapter.class);
51
52
53 private final String id;
54
55
56 private final String tableName;
57
58
59
60
61 private final ReadWriteLock readWriteLock = new DummyReadWriteLock();
62
63
64 private static volatile IgniteClient sharedClient;
65
66
67 private final IgniteClient client;
68
69
70 private final KeyValueView<Tuple, Tuple> cache;
71
72
73 static final String DEFAULT_ADDRESSES = "127.0.0.1:10800";
74
75
76 static final String CFG_PATH = "config/default-config.properties";
77
78
79 static final String KEY_COL = "key";
80
81
82 static final String VAL_COL = "val";
83
84
85
86
87 private static IgniteClient getOrCreateIgniteClient() {
88 if (sharedClient == null) {
89 synchronized (IgniteCacheAdapter.class) {
90 if (sharedClient == null) {
91 sharedClient = createIgniteClient();
92 }
93 }
94 }
95 return sharedClient;
96 }
97
98
99
100
101 static IgniteClient createIgniteClient() {
102 String addresses = DEFAULT_ADDRESSES;
103 Properties props = new Properties();
104 try (InputStream is = Files.newInputStream(Path.of(CFG_PATH))) {
105 props.load(is);
106 addresses = props.getProperty("ignite.addresses", DEFAULT_ADDRESSES);
107 } catch (IOException e) {
108 log.debug("Ignite config file not found at '" + CFG_PATH + "', using defaults.");
109 log.trace("" + e);
110 }
111 return IgniteClient.builder().addresses(addresses.split(",")).build();
112 }
113
114
115
116
117
118
119
120 public IgniteCacheAdapter(String id) {
121 this(requireNonNullId(id), getOrCreateIgniteClient());
122 }
123
124 private static String requireNonNullId(String id) {
125 if (id == null) {
126 throw new IllegalArgumentException("Cache instances require an ID");
127 }
128 return id;
129 }
130
131
132
133
134
135
136
137
138
139
140 IgniteCacheAdapter(String id, IgniteClient igniteClient) {
141 this.id = requireNonNullId(id);
142 this.tableName = toTableName(id);
143 this.client = igniteClient;
144
145 igniteClient.catalog().createTable(
146 TableDefinition.builder(tableName).ifNotExists().columns(ColumnDefinition.column(KEY_COL, ColumnType.VARBINARY),
147 ColumnDefinition.column(VAL_COL, ColumnType.VARBINARY)).primaryKey(KEY_COL).build());
148
149 cache = igniteClient.tables().table(tableName).keyValueView();
150 }
151
152 @Override
153 public String getId() {
154 return this.id;
155 }
156
157 @Override
158 public void putObject(Object key, Object value) {
159 cache.put(null, Tuple.create().set(KEY_COL, serialize(key)), Tuple.create().set(VAL_COL, serialize(value)));
160 }
161
162 @Override
163 public Object getObject(Object key) {
164 Tuple valueTuple = cache.get(null, Tuple.create().set(KEY_COL, serialize(key)));
165 return valueTuple != null ? deserialize(valueTuple.bytesValue(VAL_COL)) : null;
166 }
167
168 @Override
169 public Object removeObject(Object key) {
170 Tuple valueTuple = cache.getAndRemove(null, Tuple.create().set(KEY_COL, serialize(key)));
171 return valueTuple != null ? deserialize(valueTuple.bytesValue(VAL_COL)) : null;
172 }
173
174 @Override
175 public void clear() {
176 cache.removeAll(null);
177 }
178
179 @Override
180 public int getSize() {
181 try (ResultSet<SqlRow> rs = client.sql().execute(null, "SELECT COUNT(*) FROM " + tableName)) {
182 return rs.hasNext() ? (int) rs.next().longValue(0) : 0;
183 }
184 }
185
186 @Override
187 public ReadWriteLock getReadWriteLock() {
188 return readWriteLock;
189 }
190
191 static String toTableName(String id) {
192
193 return id.replaceAll("[^a-zA-Z0-9_]", "_").toUpperCase();
194 }
195
196 static byte[] serialize(Object obj) {
197 try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
198 ObjectOutputStream oos = new ObjectOutputStream(baos)) {
199 oos.writeObject(obj);
200 return baos.toByteArray();
201 } catch (IOException e) {
202 throw new IllegalArgumentException("Cannot serialize object of type " + obj.getClass().getName(), e);
203 }
204 }
205
206 static Object deserialize(byte[] bytes) {
207 if (bytes == null) {
208 return null;
209 }
210 try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
211 ObjectInputStream ois = new ObjectInputStream(bais)) {
212 return ois.readObject();
213 } catch (IOException | ClassNotFoundException e) {
214 throw new IllegalStateException("Cannot deserialize cache object", e);
215 }
216 }
217 }