[infinispan-commits] Infinispan SVN: r1329 - in trunk/server/memcached/src: test/java/org/infinispan/server/memcached and 1 other directory.
infinispan-commits at lists.jboss.org
infinispan-commits at lists.jboss.org
Mon Dec 28 09:18:24 EST 2009
Author: galder.zamarreno at jboss.com
Date: 2009-12-28 09:18:24 -0500 (Mon, 28 Dec 2009)
New Revision: 1329
Added:
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/FlushAllCommand.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/QuitCommand.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/VersionCommand.java
Modified:
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/AbstractVisitor.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/CommandFactory.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/MemcachedTextServer.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Reply.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextCommandDecoder.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextProtocolPipelineFactory.java
trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Visitor.java
trunk/server/memcached/src/test/java/org/infinispan/server/memcached/FunctionalTest.java
Log:
[ISPN-173] (Build memcached server module) Flush all, quit and version commands completed.
Modified: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/AbstractVisitor.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/AbstractVisitor.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/AbstractVisitor.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -92,6 +92,21 @@
return handleDefault(ch, command);
}
+ @Override
+ public Object visitFlushAll(Channel ch, FlushAllCommand command) throws Exception {
+ return handleDefault(ch, command);
+ }
+
+ @Override
+ public Object visitVersion(Channel ch, VersionCommand command) throws Exception {
+ return handleDefault(ch, command);
+ }
+
+ @Override
+ public Object visitQuit(Channel ch, QuitCommand command) throws Exception {
+ return handleDefault(ch, command);
+ }
+
protected Object handleDefault(Channel ch, Command command) throws Exception {
return null;
}
Modified: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/CommandFactory.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/CommandFactory.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/CommandFactory.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -29,6 +29,9 @@
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.infinispan.Cache;
@@ -47,10 +50,12 @@
private final Cache cache;
private final InterceptorChain chain;
+ private final ScheduledExecutorService scheduler;
- public CommandFactory(Cache cache, InterceptorChain chain) {
+ public CommandFactory(Cache cache, InterceptorChain chain, ScheduledExecutorService scheduler) {
this.cache = cache;
this.chain = chain;
+ this.scheduler = scheduler;
}
public Command createCommand(String line) throws IOException {
@@ -92,6 +97,13 @@
return NumericCommand.newNumericCommand(cache, type, key, value);
case STATS:
return StatsCommand.newStatsCommand(cache, type, chain);
+ case FLUSH_ALL:
+ long delay = args.length > 1 ? Long.parseLong(args[1]) : 0;
+ return FlushAllCommand.newFlushAllCommand(cache, delay, scheduler);
+ case VERSION:
+ return VersionCommand.newVersionCommand();
+ case QUIT:
+ return QuitCommand.newQuitCommand();
default:
throw new NotImplementedException("Parsed type not implemented yet");
}
Added: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/FlushAllCommand.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/FlushAllCommand.java (rev 0)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/FlushAllCommand.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -0,0 +1,91 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2009, Red Hat, Inc. and/or its affiliates, and
+ * individual contributors as indicated by the @author tags. See the
+ * copyright.txt file in the distribution for a full listing of
+ * individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.infinispan.server.memcached;
+
+import static org.infinispan.server.memcached.Reply.OK;
+import static org.infinispan.server.memcached.TextProtocolUtil.CRLF;
+import static org.jboss.netty.buffer.ChannelBuffers.wrappedBuffer;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.infinispan.Cache;
+import org.infinispan.context.Flag;
+import org.jboss.netty.channel.Channel;
+
+/**
+ * FlushAllCommand.
+ *
+ * @author Galder Zamarreño
+ * @since 4.0
+ */
+public class FlushAllCommand implements Command {
+ final Cache cache;
+ final long delay;
+ final ScheduledExecutorService scheduler;
+
+ FlushAllCommand(Cache cache, long delay, ScheduledExecutorService scheduler) {
+ this.cache = cache;
+ this.delay = delay;
+ this.scheduler = scheduler;
+ }
+
+ @Override
+ public Object acceptVisitor(Channel ch, CommandInterceptor next) throws Exception {
+ return next.visitFlushAll(ch, this);
+ }
+
+ @Override
+ public CommandType getType() {
+ return CommandType.FLUSH_ALL;
+ }
+
+ @Override
+ public Object perform(Channel ch) throws Exception {
+ if (delay == 0) {
+ cache.getAdvancedCache().withFlags(Flag.CACHE_MODE_LOCAL, Flag.SKIP_CACHE_STORE).clear();
+ } else {
+ scheduler.schedule(new FlushAllDelayed(cache), delay, TimeUnit.SECONDS);
+ }
+ ch.write(wrappedBuffer(wrappedBuffer(OK.toString().getBytes()), wrappedBuffer(CRLF)));
+ return null;
+ }
+
+ public static FlushAllCommand newFlushAllCommand(Cache cache, long delay, ScheduledExecutorService scheduler) {
+ return new FlushAllCommand(cache, delay, scheduler);
+ }
+
+ private static class FlushAllDelayed implements Runnable {
+ final Cache cache;
+
+ FlushAllDelayed(Cache cache) {
+ this.cache = cache;
+ }
+
+ @Override
+ public void run() {
+ cache.getAdvancedCache().withFlags(Flag.CACHE_MODE_LOCAL, Flag.SKIP_CACHE_STORE).clear();
+ }
+ }
+
+}
Modified: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/MemcachedTextServer.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/MemcachedTextServer.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/MemcachedTextServer.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -23,7 +23,11 @@
package org.infinispan.server.memcached;
import java.net.InetSocketAddress;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import org.infinispan.Cache;
import org.infinispan.manager.CacheManager;
@@ -40,10 +44,12 @@
public class MemcachedTextServer {
private final CacheManager manager;
private final int port;
+ private final ScheduledExecutorService scheduler;
public MemcachedTextServer(CacheManager manager, int port) {
this.manager = manager;
this.port = port;
+ this.scheduler = Executors.newScheduledThreadPool(1);
}
public int getPort() {
@@ -58,11 +64,12 @@
ChannelFactory factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
ServerBootstrap bootstrap = new ServerBootstrap(factory);
InterceptorChain chain = InterceptorChainFactory.getInstance(cache).buildInterceptorChain();
- bootstrap.setPipelineFactory(new TextProtocolPipelineFactory(cache, chain));
+ bootstrap.setPipelineFactory(new TextProtocolPipelineFactory(cache, chain, scheduler));
bootstrap.bind(new InetSocketAddress(port));
}
public void stop() {
manager.stop();
+ scheduler.shutdown();
}
}
Added: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/QuitCommand.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/QuitCommand.java (rev 0)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/QuitCommand.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -0,0 +1,55 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2009, Red Hat, Inc. and/or its affiliates, and
+ * individual contributors as indicated by the @author tags. See the
+ * copyright.txt file in the distribution for a full listing of
+ * individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.infinispan.server.memcached;
+
+import org.jboss.netty.channel.Channel;
+
+/**
+ * QuitCommand.
+ *
+ * @author Galder Zamarreño
+ * @since 4.0
+ */
+public enum QuitCommand implements Command {
+ INSTANCE;
+
+ @Override
+ public Object acceptVisitor(Channel ch, CommandInterceptor next) throws Exception {
+ return next.visitQuit(ch, this);
+ }
+
+ @Override
+ public CommandType getType() {
+ return CommandType.QUIT;
+ }
+
+ @Override
+ public Object perform(Channel ch) throws Exception {
+ ch.disconnect();
+ return null;
+ }
+
+ public static QuitCommand newQuitCommand() {
+ return INSTANCE;
+ }
+}
Modified: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Reply.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Reply.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Reply.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -29,7 +29,7 @@
* @since 4.0
*/
public enum Reply {
- STORED, NOT_STORED, EXISTS, NOT_FOUND, DELETED, STAT, VALUE, END,
+ STORED, NOT_STORED, EXISTS, NOT_FOUND, DELETED, STAT, VALUE, END, OK, VERSION,
ERROR, CLIENT_ERROR, SERVER_ERROR;
public byte[] bytes() {
Modified: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextCommandDecoder.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextCommandDecoder.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextCommandDecoder.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -24,8 +24,7 @@
import java.io.IOException;
import java.io.StreamCorruptedException;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.ScheduledExecutorService;
import org.infinispan.Cache;
import org.infinispan.util.logging.Log;
@@ -49,15 +48,14 @@
private final CommandFactory factory;
private volatile Command command;
-// private final AtomicBoolean corrupted = new AtomicBoolean();
protected enum State {
READ_COMMAND, READ_UNSTRUCTURED_DATA;
}
- TextCommandDecoder(Cache cache, InterceptorChain chain) {
+ TextCommandDecoder(Cache cache, InterceptorChain chain, ScheduledExecutorService scheduler) {
super(State.READ_COMMAND, true);
- factory = new CommandFactory(cache, chain);
+ factory = new CommandFactory(cache, chain, scheduler);
}
@Override
Modified: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextProtocolPipelineFactory.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextProtocolPipelineFactory.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/TextProtocolPipelineFactory.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -25,6 +25,7 @@
import static org.jboss.netty.channel.Channels.*;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
import org.infinispan.Cache;
import org.jboss.netty.channel.ChannelHandler;
@@ -51,8 +52,8 @@
private final ReplayingDecoder<TextCommandDecoder.State> decoder;
private final ChannelHandler handler;
- public TextProtocolPipelineFactory(Cache cache, InterceptorChain chain) {
- this.decoder = new TextCommandDecoder(cache, chain);
+ public TextProtocolPipelineFactory(Cache cache, InterceptorChain chain, ScheduledExecutorService scheduler) {
+ this.decoder = new TextCommandDecoder(cache, chain, scheduler);
this.handler = new TextCommandHandler(cache, chain);
}
Added: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/VersionCommand.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/VersionCommand.java (rev 0)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/VersionCommand.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -0,0 +1,61 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2009, Red Hat, Inc. and/or its affiliates, and
+ * individual contributors as indicated by the @author tags. See the
+ * copyright.txt file in the distribution for a full listing of
+ * individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.infinispan.server.memcached;
+
+import static org.infinispan.server.memcached.Reply.VERSION;
+import static org.infinispan.server.memcached.TextProtocolUtil.CRLF;
+import static org.jboss.netty.buffer.ChannelBuffers.wrappedBuffer;
+
+import org.infinispan.Version;
+import org.jboss.netty.channel.Channel;
+
+/**
+ * VersionCommand.
+ *
+ * @author Galder Zamarreño
+ * @since 4.0
+ */
+public enum VersionCommand implements Command {
+ INSTANCE;
+
+ @Override
+ public Object acceptVisitor(Channel ch, CommandInterceptor next) throws Exception {
+ return next.visitVersion(ch, this);
+ }
+
+ @Override
+ public CommandType getType() {
+ return CommandType.VERSION;
+ }
+
+ @Override
+ public Object perform(Channel ch) throws Exception {
+ String version = ' ' + Version.version;
+ ch.write(wrappedBuffer(wrappedBuffer(VERSION.bytes()), wrappedBuffer(version.getBytes()), wrappedBuffer(CRLF)));
+ return null;
+ }
+
+ public static VersionCommand newVersionCommand() {
+ return INSTANCE;
+ }
+}
Modified: trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Visitor.java
===================================================================
--- trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Visitor.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/main/java/org/infinispan/server/memcached/Visitor.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -43,7 +43,7 @@
Object visitIncrement(Channel ch, IncrementCommand command) throws Exception;
Object visitDecrement(Channel ch, DecrementCommand command) throws Exception;
Object visitStats(Channel ch, StatsCommand command) throws Exception;
-/// Object visitFlushAll(Channel ch, FlushAllCommand command);
-// Object visitVersion(Channel ch, VersionCommand command);
-// Object visitQuit(Channel ch, QuitCommand command);
+ Object visitFlushAll(Channel ch, FlushAllCommand command) throws Exception;
+ Object visitVersion(Channel ch, VersionCommand command) throws Exception;
+ Object visitQuit(Channel ch, QuitCommand command) throws Exception;
}
Modified: trunk/server/memcached/src/test/java/org/infinispan/server/memcached/FunctionalTest.java
===================================================================
--- trunk/server/memcached/src/test/java/org/infinispan/server/memcached/FunctionalTest.java 2009-12-28 11:34:15 UTC (rev 1328)
+++ trunk/server/memcached/src/test/java/org/infinispan/server/memcached/FunctionalTest.java 2009-12-28 14:18:24 UTC (rev 1329)
@@ -23,6 +23,7 @@
package org.infinispan.server.memcached;
import java.lang.reflect.Method;
+import java.net.SocketAddress;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.Future;
@@ -32,9 +33,11 @@
import net.spy.memcached.CASValue;
import net.spy.memcached.MemcachedClient;
+import org.infinispan.Version;
import org.infinispan.manager.CacheManager;
import org.infinispan.server.memcached.test.MemcachedTestingUtil;
import org.infinispan.test.SingleCacheManagerTest;
+import org.infinispan.test.TestingUtil;
import org.infinispan.test.fwk.TestCacheManagerFactory;
import org.testng.annotations.AfterClass;
import org.testng.annotations.Test;
@@ -322,4 +325,50 @@
assert 0 == newValue : "Unexpected result: " + newValue;
}
+ public void testFlushAll(Method m) throws Exception {
+ Future<Boolean> f;
+ for (int i = 0; i < 5; i++) {
+ String key = k(m, "k" + i + "-");
+ Object value = v(m, "v" + i + "-");
+ f = client.set(key, 0, value);
+ assert f.get(5, TimeUnit.SECONDS);
+ assert value.equals(client.get(key));
+ }
+
+ f = client.flush();
+ assert f.get(5, TimeUnit.SECONDS);
+
+ for (int i = 0; i < 5; i++) {
+ String key = k(m, "k" + i + "-");
+ assert null == client.get(key);
+ }
+ }
+
+ public void testFlushAllDelayed(Method m) throws Exception {
+ Future<Boolean> f;
+ for (int i = 0; i < 5; i++) {
+ String key = k(m, "k" + i + "-");
+ Object value = v(m, "v" + i + "-");
+ f = client.set(key, 0, value);
+ assert f.get(5, TimeUnit.SECONDS);
+ assert value.equals(client.get(key));
+ }
+
+ f = client.flush(2);
+ assert f.get(5, TimeUnit.SECONDS);
+
+ TestingUtil.sleepThread(2200);
+
+ for (int i = 0; i < 5; i++) {
+ String key = k(m, "k" + i + "-");
+ assert null == client.get(key);
+ }
+ }
+
+ public void testVersion() throws Exception {
+ Map<SocketAddress, String> versions = client.getVersions();
+ assert 1 == versions.size();
+ String version = versions.values().iterator().next();
+ assert Version.version.equals(version);
+ }
}
More information about the infinispan-commits
mailing list