Author: frm
Date: Fri Sep 9 17:12:13 2016
New Revision: 1760082
URL: http://svn.apache.org/viewvc?rev=1760082&view=rev
Log:
OAK-4785 - Implement request parsing in a new handler
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequest.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequest.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequest.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoder.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoderTest.java
(with props)
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/SegmentLoaderHandler.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClientHandler.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/Messages.java
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/SegmentLoaderHandler.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/SegmentLoaderHandler.java?rev=1760082&r1=1760081&r2=1760082&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/SegmentLoaderHandler.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/SegmentLoaderHandler.java
Fri Sep 9 17:12:13 2016
@@ -19,8 +19,8 @@
package org.apache.jackrabbit.oak.segment.standby.client;
import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;
-import static
org.apache.jackrabbit.oak.segment.standby.codec.Messages.newGetBlobReq;
-import static
org.apache.jackrabbit.oak.segment.standby.codec.Messages.newGetSegmentReq;
+import static
org.apache.jackrabbit.oak.segment.standby.codec.Messages.newGetBlobRequest;
+import static
org.apache.jackrabbit.oak.segment.standby.codec.Messages.newGetSegmentRequest;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
@@ -180,7 +180,7 @@ public class SegmentLoaderHandler extend
@Override
public Segment readSegment(final String id) {
// Use the I/O thread to write the request to the server
- ctx.writeAndFlush(newGetSegmentReq(this.clientID, id));
+ ctx.writeAndFlush(newGetSegmentRequest(this.clientID, id));
// Wait on the sync thread for the response.
return getSegment(id);
}
@@ -188,7 +188,7 @@ public class SegmentLoaderHandler extend
@Override
public Blob readBlob(String blobId) {
// Use the I/O thread to write the request to the server
- ctx.writeAndFlush(newGetBlobReq(this.clientID, blobId));
+ ctx.writeAndFlush(newGetBlobRequest(this.clientID, blobId));
// Wait on the sync thread for the response.
return getBlob(blobId);
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClientHandler.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClientHandler.java?rev=1760082&r1=1760081&r2=1760082&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClientHandler.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClientHandler.java
Fri Sep 9 17:12:13 2016
@@ -19,7 +19,7 @@
package org.apache.jackrabbit.oak.segment.standby.client;
-import static
org.apache.jackrabbit.oak.segment.standby.codec.Messages.newGetHeadReq;
+import static
org.apache.jackrabbit.oak.segment.standby.codec.Messages.newGetHeadRequest;
import java.io.Closeable;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -57,7 +57,7 @@ public class StandbyClientHandler extend
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("sending head request");
- ctx.writeAndFlush(newGetHeadReq(this.observer.getID()));
+ ctx.writeAndFlush(newGetHeadRequest(this.observer.getID()));
log.debug("did send head request");
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/Messages.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/Messages.java?rev=1760082&r1=1760081&r2=1760082&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/Messages.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/Messages.java
Fri Sep 9 17:12:13 2016
@@ -31,20 +31,45 @@ public class Messages {
private static final String MAGIC = "Standby-CMD@";
private static final String SEPARATOR = ":";
- private static String newRequest(String clientID, String body) {
- return MAGIC + (clientID == null ? "" : clientID.replace(SEPARATOR,
"#")) + SEPARATOR + body + "\r\n";
+ private static String newRequest(String clientId, String body, boolean
delimited) {
+ StringBuilder builder = new StringBuilder(MAGIC);
+
+ if (clientId != null) {
+ builder.append(clientId.replace(SEPARATOR, "#"));
+ }
+
+ builder.append(SEPARATOR);
+ builder.append(body);
+
+ if (delimited) {
+ builder.append("\r\n");
+ }
+
+ return builder.toString();
+ }
+
+ public static String newGetHeadRequest(String clientId, boolean delimited)
{
+ return newRequest(clientId, GET_HEAD, delimited);
+ }
+
+ public static String newGetHeadRequest(String clientId) {
+ return newGetHeadRequest(clientId, true);
+ }
+
+ public static String newGetSegmentRequest(String clientId, String
segmentId, boolean delimited) {
+ return newRequest(clientId, GET_SEGMENT + segmentId, delimited);
}
- public static String newGetHeadReq(String clientID) {
- return newRequest(clientID, GET_HEAD);
+ public static String newGetSegmentRequest(String clientId, String
segmentId) {
+ return newGetSegmentRequest(clientId, segmentId, true);
}
- public static String newGetSegmentReq(String clientID, String sid) {
- return newRequest(clientID, GET_SEGMENT + sid);
+ public static String newGetBlobRequest(String clientId, String blobId,
boolean delimited) {
+ return newRequest(clientId, GET_BLOB + blobId, delimited);
}
- public static String newGetBlobReq(String clientID, String blobId) {
- return newRequest(clientID, GET_BLOB + blobId);
+ public static String newGetBlobRequest(String clientId, String blobId) {
+ return newGetBlobRequest(clientId, blobId, true);
}
public static String extractMessageFrom(String payload) {
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequest.java?rev=1760082&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequest.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequest.java
Fri Sep 9 17:12:13 2016
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.jackrabbit.oak.segment.standby.server;
+
+class GetBlobRequest {
+
+ private final String clientId;
+
+ private final String blobId;
+
+ GetBlobRequest(String clientId, String blobId) {
+ this.clientId = clientId;
+ this.blobId = blobId;
+ }
+
+ String getClientId() {
+ return clientId;
+ }
+
+ String getBlobId() {
+ return blobId;
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequest.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequest.java?rev=1760082&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequest.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequest.java
Fri Sep 9 17:12:13 2016
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.jackrabbit.oak.segment.standby.server;
+
+class GetHeadRequest {
+
+ private String clientId;
+
+ GetHeadRequest(String clientId) {
+ this.clientId = clientId;
+ }
+
+ String getClientId() {
+ return clientId;
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequest.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequest.java?rev=1760082&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequest.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequest.java
Fri Sep 9 17:12:13 2016
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.jackrabbit.oak.segment.standby.server;
+
+class GetSegmentRequest {
+
+ private final String clientId;
+
+ private final String segmentId;
+
+ GetSegmentRequest(String clientId, String segmentId) {
+ this.clientId = clientId;
+ this.segmentId = segmentId;
+ }
+
+ public String getClientId() {
+ return clientId;
+ }
+
+ public String getSegmentId() {
+ return segmentId;
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequest.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoder.java?rev=1760082&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoder.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoder.java
Fri Sep 9 17:12:13 2016
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.jackrabbit.oak.segment.standby.server;
+
+import java.util.List;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import org.apache.jackrabbit.oak.segment.standby.codec.Messages;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Implements the logic for parsing input messages and failing when an
incorrect
+ * or unrecognized message is received.
+ */
+class RequestDecoder extends MessageToMessageDecoder<String> {
+
+ private static final Logger log =
LoggerFactory.getLogger(RequestDecoder.class);
+
+ @Override
+ protected void decode(ChannelHandlerContext ctx, String msg, List<Object>
out) throws Exception {
+ String request = Messages.extractMessageFrom(msg);
+
+ if (request == null) {
+ log.debug("Unable to parse message '{}'", msg);
+ ctx.close();
+ } else if (request.equalsIgnoreCase(Messages.GET_HEAD)) {
+ log.debug("Parsed 'get head' message");
+ out.add(new GetHeadRequest(Messages.extractClientFrom(msg)));
+ } else if (request.startsWith(Messages.GET_SEGMENT)) {
+ log.debug("Parsed 'get segment' message");
+ out.add(new GetSegmentRequest(Messages.extractClientFrom(msg),
request.substring(Messages.GET_SEGMENT.length())));
+ } else if (request.startsWith(Messages.GET_BLOB)) {
+ log.debug("Parsed 'get blob' message");
+ out.add(new GetBlobRequest(Messages.extractClientFrom(msg),
request.substring(Messages.GET_BLOB.length())));
+ } else {
+ log.debug("Unrecognized message '{}'", msg);
+ ctx.close();
+ }
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoder.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoderTest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoderTest.java?rev=1760082&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoderTest.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoderTest.java
Fri Sep 9 17:12:13 2016
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.jackrabbit.oak.segment.standby.server;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.jackrabbit.oak.segment.standby.codec.Messages;
+import org.junit.Test;
+
+public class RequestDecoderTest {
+
+ @Test
+ public void shouldDecodeGetHeadRequests() throws Exception {
+ EmbeddedChannel channel = new EmbeddedChannel(new RequestDecoder());
+ channel.writeInbound(Messages.newGetHeadRequest("clientId", false));
+ GetHeadRequest request = (GetHeadRequest) channel.readInbound();
+ assertEquals("clientId", request.getClientId());
+ }
+
+ @Test
+ public void shouldDecodeGetSegmentRequests() throws Exception {
+ EmbeddedChannel channel = new EmbeddedChannel(new RequestDecoder());
+ channel.writeInbound(Messages.newGetSegmentRequest("clientId",
"segmentId", false));
+ GetSegmentRequest request = (GetSegmentRequest) channel.readInbound();
+ assertEquals("clientId", request.getClientId());
+ assertEquals("segmentId", request.getSegmentId());
+ }
+
+ @Test
+ public void shouldDecodeGetBlobRequests() throws Exception {
+ EmbeddedChannel channel = new EmbeddedChannel(new RequestDecoder());
+ channel.writeInbound(Messages.newGetBlobRequest("clientId", "blobId",
false));
+ GetBlobRequest request = (GetBlobRequest) channel.readInbound();
+ assertEquals("clientId", request.getClientId());
+ assertEquals("blobId", request.getBlobId());
+ }
+
+ @Test
+ public void shouldCloseConnectionOnGarbage() throws Exception {
+ EmbeddedChannel channel = new EmbeddedChannel(new RequestDecoder());
+ channel.writeInbound("this is garbage");
+ assertFalse(channel.isOpen());
+ }
+
+ @Test
+ public void shouldCloseConnectionOnUnrecognizedMessageTypes() throws
Exception {
+ EmbeddedChannel channel = new EmbeddedChannel(new RequestDecoder());
+ channel.writeInbound("Standby-CMD@clientId:z");
+ assertFalse(channel.isOpen());
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/RequestDecoderTest.java
------------------------------------------------------------------------------
svn:eol-style = native
