honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Memcached transport does not support version, npe when no uri, closes #342.

This commit is contained in:
kimchy 2010-09-01 15:16:03 +03:00
parent 38cae95ebc
commit 46b58ce77f
1 changed files with 40 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
plugins/transport/memcached/src/main/java/org/elasticsearch/memcached/netty/MemcachedDecoder.java
View File

@ -19,6 +19,7 @@
package org.elasticsearch.memcached.netty; package org.elasticsearch.memcached.netty;
import org.elasticsearch.Version;
import org.elasticsearch.common.Unicode; import org.elasticsearch.common.Unicode;
import org.elasticsearch.common.logging.ESLogger; import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.netty.buffer.ChannelBuffer; import org.elasticsearch.common.netty.buffer.ChannelBuffer;
@ -49,6 +50,7 @@ public class MemcachedDecoder extends FrameDecoder {
private volatile StringBuffer sb = new StringBuffer(); private volatile StringBuffer sb = new StringBuffer();
private volatile MemcachedRestRequest request; private volatile MemcachedRestRequest request;
private volatile boolean ending = false;
public MemcachedDecoder(ESLogger logger) { public MemcachedDecoder(ESLogger logger) {
super(false); super(false);
@ -65,7 +67,10 @@ public class MemcachedDecoder extends FrameDecoder {
} }
short magic = buffer.readUnsignedByte(); short magic = buffer.readUnsignedByte();
if (magic == 0x80) { if (magic == 0x80) {
if (buffer.readableBytes() < 23) return null; if (buffer.readableBytes() < 23) {
buffer.resetReaderIndex(); // but back magic
return null;
}
short opcode = buffer.readUnsignedByte(); short opcode = buffer.readUnsignedByte();
short keyLength = buffer.readShort(); short keyLength = buffer.readShort();
short extraLength = buffer.readUnsignedByte(); short extraLength = buffer.readUnsignedByte();
@ -139,22 +144,23 @@ public class MemcachedDecoder extends FrameDecoder {
int readableBytes = buffer.readableBytes(); int readableBytes = buffer.readableBytes();
for (int i = 0; i < readableBytes; i++) { for (int i = 0; i < readableBytes; i++) {
byte next = buffer.readByte(); byte next = buffer.readByte();
if (next == CR) { if (!ending && next == CR) {
next = buffer.readByte(); ending = true;
if (next == LF) { } else if (ending && next == LF) {
done = true; ending = false;
break;
}
} else if (next == LF) {
done = true; done = true;
break; break;
} else if (ending) {
logger.error("Corrupt stream, expected LF, found [0x{}]", Integer.toHexString(next));
throw new StreamCorruptedException("Expecting LF after CR");
} else { } else {
sb.append((char) next); sb.append((char) next);
} }
} }
if (!done) { if (!done) {
sb.setLength(0); // let's keep the buffer and bytes read
buffer.resetReaderIndex(); // buffer.discardReadBytes();
buffer.markReaderIndex();
return null; return null;
} }
@ -163,27 +169,36 @@ public class MemcachedDecoder extends FrameDecoder {
sb.setLength(0); sb.setLength(0);
String cmd = args[0]; String cmd = args[0];
String uri = args[1];
if ("get".equals(cmd)) { if ("get".equals(cmd)) {
request = new MemcachedRestRequest(RestRequest.Method.GET, uri, null, -1, false); request = new MemcachedRestRequest(RestRequest.Method.GET, args[1], null, -1, false);
if (args.length > 3) { if (args.length > 3) {
request.setData(Unicode.fromStringAsBytes(args[2])); request.setData(Unicode.fromStringAsBytes(args[2]));
} }
return request; return request;
} else if ("delete".equals(cmd)) { } else if ("delete".equals(cmd)) {
request = new MemcachedRestRequest(RestRequest.Method.DELETE, uri, null, -1, false); request = new MemcachedRestRequest(RestRequest.Method.DELETE, args[1], null, -1, false);
// if (args.length > 3) { // if (args.length > 3) {
// request.setData(Unicode.fromStringAsBytes(args[2])); // request.setData(Unicode.fromStringAsBytes(args[2]));
// } // }
return request; return request;
} else if ("set".equals(cmd)) { } else if ("set".equals(cmd)) {
this.request = new MemcachedRestRequest(RestRequest.Method.POST, uri, null, Integer.parseInt(args[4]), false); this.request = new MemcachedRestRequest(RestRequest.Method.POST, args[1], null, Integer.parseInt(args[4]), false);
buffer.markReaderIndex(); buffer.markReaderIndex();
} else if ("version".equals(cmd)) { // sent as a noop
byte[] bytes = Version.full().getBytes();
ChannelBuffer writeBuffer = ChannelBuffers.dynamicBuffer(bytes.length);
writeBuffer.writeBytes(bytes);
channel.write(writeBuffer);
return MemcachedDispatcher.IGNORE_REQUEST;
} else if ("quit".equals(cmd)) { } else if ("quit".equals(cmd)) {
if (channel.isConnected()) { // we maybe in the process of clearing the queued bits
channel.disconnect(); channel.disconnect();
}
} else { } else {
logger.error("Unsupported command [{}], ignoring and closing connection", cmd); logger.error("Unsupported command [{}], ignoring and closing connection", cmd);
if (channel.isConnected()) { // we maybe in the process of clearing the queued bits
channel.disconnect(); channel.disconnect();
}
return null; return null;
} }
} }
@ -215,7 +230,13 @@ public class MemcachedDecoder extends FrameDecoder {
@Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { @Override public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
this.request = null; this.request = null;
sb.setLength(0); this.ending = false;
e.getCause().printStackTrace(); this.sb.setLength(0);
if (ctx.getChannel().isConnected()) {
ctx.getChannel().disconnect();
}
logger.error("caught exception on memcached decoder", e);
} }
} }