/**
* Copyright (C) 2012-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package ninja.cache;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamClass;
import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import net.spy.memcached.AddrUtil;
import net.spy.memcached.ConnectionFactory;
import net.spy.memcached.ConnectionFactoryBuilder;
import net.spy.memcached.MemcachedClient;
import net.spy.memcached.auth.AuthDescriptor;
import net.spy.memcached.auth.PlainCallbackHandler;
import net.spy.memcached.transcoders.SerializingTranscoder;
import ninja.lifecycle.Dispose;
import ninja.utils.NinjaConstant;
import ninja.utils.NinjaProperties;
import org.slf4j.Logger;
import com.google.inject.Inject;
import com.google.inject.Singleton;
/**
* Memcached implementation (using http://code.google.com/p/spymemcached/)
* expiration is specified in seconds.
*
* Heavily inspired by excellent Play! 1.2.5 implementation.
*/
@Singleton
public class CacheMemcachedImpl implements Cache {
private final Logger logger;
private final MemcachedClient client;
private final SerializingTranscoder tc;
private final NinjaProperties ninjaProperties;
@Inject
public CacheMemcachedImpl(
final Logger logger,
final NinjaProperties ninjaProperties) throws Exception {
this.logger = logger;
this.ninjaProperties = ninjaProperties;
this.tc = new SerializingTranscoder() {
@Override
protected Object deserialize(byte[] data) {
try {
return new ObjectInputStream(new ByteArrayInputStream(data)) {
@Override
protected Class<?> resolveClass(ObjectStreamClass desc)
throws IOException, ClassNotFoundException {
return Class.forName(desc.getName(), false, Thread.currentThread().getContextClassLoader());
}
}.readObject();
} catch (Exception e) {
logger.error("Could not deserialize", e);
}
return null;
}
@Override
protected byte[] serialize(Object object) {
try {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
new ObjectOutputStream(bos).writeObject(object);
return bos.toByteArray();
} catch (IOException e) {
logger.error("Could not serialize", e);
}
return null;
}
};
System.setProperty("net.spy.log.LoggerImpl", "net.spy.memcached.compat.log.Log4JLogger");
List<InetSocketAddress> addrs;
String allMemcachedHosts = ninjaProperties.getOrDie(NinjaConstant.MEMCACHED_HOST);
addrs = AddrUtil.getAddresses(allMemcachedHosts);
String memcachedUser = ninjaProperties.get(NinjaConstant.MEMCACHED_USER);
if (memcachedUser != null) {
String memcachePassword = ninjaProperties.getOrDie(NinjaConstant.MEMCACHED_PASSWORD);
// Use plain SASL to connect to memcached
AuthDescriptor ad = new AuthDescriptor(new String[]{"PLAIN"},
new PlainCallbackHandler(memcachedUser, memcachePassword));
ConnectionFactory cf = new ConnectionFactoryBuilder()
.setProtocol(ConnectionFactoryBuilder.Protocol.BINARY)
.setAuthDescriptor(ad)
.build();
client = new MemcachedClient(cf, addrs);
} else {
client = new MemcachedClient(addrs);
}
}
public void add(String key, Object value, int expiration) {
client.add(key, expiration, value, tc);
}
public Object get(String key) {
Future<Object> future = client.asyncGet(key, tc);
try {
return future.get(1, TimeUnit.SECONDS);
} catch (Exception e) {
future.cancel(false);
}
return null;
}
public void clear() {
client.flush();
}
public void delete(String key) {
client.delete(key);
}
public Map<String, Object> get(String[] keys) {
Future<Map<String, Object>> future = client.asyncGetBulk(tc, keys);
try {
return future.get(1, TimeUnit.SECONDS);
} catch (Exception e) {
future.cancel(false);
}
return Collections.<String, Object>emptyMap();
}
public long incr(String key, int by) {
return client.incr(key, by, 0);
}
public long decr(String key, int by) {
return client.decr(key, by, 0);
}
public void replace(String key, Object value, int expiration) {
client.replace(key, expiration, value, tc);
}
public boolean safeAdd(String key, Object value, int expiration) {
Future<Boolean> future = client.add(key, expiration, value, tc);
try {
return future.get(1, TimeUnit.SECONDS);
} catch (Exception e) {
future.cancel(false);
}
return false;
}
public boolean safeDelete(String key) {
Future<Boolean> future = client.delete(key);
try {
return future.get(1, TimeUnit.SECONDS);
} catch (Exception e) {
future.cancel(false);
}
return false;
}
public boolean safeReplace(String key, Object value, int expiration) {
Future<Boolean> future = client.replace(key, expiration, value, tc);
try {
return future.get(1, TimeUnit.SECONDS);
} catch (Exception e) {
future.cancel(false);
}
return false;
}
public boolean safeSet(String key, Object value, int expiration) {
Future<Boolean> future = client.set(key, expiration, value, tc);
try {
return future.get(1, TimeUnit.SECONDS);
} catch (Exception e) {
future.cancel(false);
}
return false;
}
public void set(String key, Object value, int expiration) {
client.set(key, expiration, value, tc);
}
@Dispose
public void stop() {
client.shutdown();
}
}