Author: frm
Date: Thu Sep 15 14:03:05 2016
New Revision: 1760934
URL: http://svn.apache.org/viewvc?rev=1760934&view=rev
Log:
OAK-4803 - Simplify the client side of the cold standby
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetBlobResponseHandler.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetHeadResponseHandler.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetSegmentResponseHandler.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyDiff.java
- copied, changed from r1760420,
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyApplyDiff.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySync.java
- copied, changed from r1760651,
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySyncExecution.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestEncoder.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestEncoder.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestEncoder.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RequestDecoder.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/ResponseDecoder.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/RequestDecoderTest.java
(with props)
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/ResponseDecoderTest.java
(with props)
Removed:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/FailedRequestListener.java
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/StandbyApplyDiff.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/BlobEncoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestDecoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestDecoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestDecoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/IdArrayBasedBlob.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RecordIdDecoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RecordIdEncoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/ReplyDecoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/SegmentEncoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/SegmentReply.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/store/RemoteSegmentLoader.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/store/StandbyStore.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestDecoderTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestDecoderTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestDecoderTest.java
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponse.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponseEncoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponse.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponseEncoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequest.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponse.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponseEncoder.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/Messages.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyBlobReader.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyHeadReader.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbySegmentReader.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequestHandler.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequestHandler.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequestHandler.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/ResponseObserverHandler.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyBlobReader.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyHeadReader.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbySegmentReader.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/StandbyServer.java
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/store/StandbyStoreService.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/BrokenNetworkTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/DataStoreTestBase.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/FailoverIPRangeTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/FailoverMultipleClientsTestIT.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/FailoverSslTestIT.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/MBeanTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/RecoverTestIT.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/StandbyTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/StandbyTestIT.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/TestBase.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/benchmark/BenchmarkBase.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/benchmark/BulkTransferBenchmark.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponseEncoderTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponseEncoderTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponseEncoderTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/GetBlobRequestHandlerTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/GetHeadRequestHandlerTest.java
jackrabbit/oak/trunk/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/standby/server/GetSegmentRequestHandlerTest.java
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetBlobResponseHandler.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetBlobResponseHandler.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetBlobResponseHandler.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetBlobResponseHandler.java
Thu Sep 15 14:03:05 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.client;
+
+import java.util.concurrent.BlockingQueue;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetBlobResponse;
+
+class GetBlobResponseHandler extends
SimpleChannelInboundHandler<GetBlobResponse> {
+
+ private final BlockingQueue<GetBlobResponse> queue;
+
+ GetBlobResponseHandler(BlockingQueue<GetBlobResponse> queue) {
+ this.queue = queue;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, GetBlobResponse
msg) throws Exception {
+ queue.offer(msg);
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetBlobResponseHandler.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetHeadResponseHandler.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetHeadResponseHandler.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetHeadResponseHandler.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetHeadResponseHandler.java
Thu Sep 15 14:03:05 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.client;
+
+import java.util.concurrent.BlockingQueue;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetHeadResponse;
+
+class GetHeadResponseHandler extends
SimpleChannelInboundHandler<GetHeadResponse> {
+
+ private final BlockingQueue<GetHeadResponse> queue;
+
+ GetHeadResponseHandler(BlockingQueue<GetHeadResponse> queue) {
+ this.queue = queue;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, GetHeadResponse
msg) throws Exception {
+ queue.offer(msg);
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetHeadResponseHandler.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetSegmentResponseHandler.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetSegmentResponseHandler.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetSegmentResponseHandler.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetSegmentResponseHandler.java
Thu Sep 15 14:03:05 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.client;
+
+import java.util.concurrent.BlockingQueue;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetSegmentResponse;
+
+class GetSegmentResponseHandler extends
SimpleChannelInboundHandler<GetSegmentResponse> {
+
+ private final BlockingQueue<GetSegmentResponse> queue;
+
+ GetSegmentResponseHandler(BlockingQueue<GetSegmentResponse> queue) {
+ this.queue = queue;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, GetSegmentResponse
msg) throws Exception {
+ queue.offer(msg);
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/GetSegmentResponseHandler.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
Thu Sep 15 14:03:05 2016
@@ -0,0 +1,180 @@
+/*
+ * 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.client;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import io.netty.handler.codec.compression.SnappyFramedDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+import io.netty.util.CharsetUtil;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetBlobRequest;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetBlobRequestEncoder;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetBlobResponse;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetHeadRequest;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetHeadRequestEncoder;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetHeadResponse;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetSegmentRequest;
+import
org.apache.jackrabbit.oak.segment.standby.codec.GetSegmentRequestEncoder;
+import org.apache.jackrabbit.oak.segment.standby.codec.GetSegmentResponse;
+import org.apache.jackrabbit.oak.segment.standby.codec.ResponseDecoder;
+
+class StandbyClient implements AutoCloseable {
+
+ private final BlockingQueue<GetHeadResponse> headQueue = new
LinkedBlockingDeque<>();
+
+ private final BlockingQueue<GetSegmentResponse> segmentQueue = new
LinkedBlockingDeque<>();
+
+ private final BlockingQueue<GetBlobResponse> blobQueue = new
LinkedBlockingDeque<>();
+
+ private final boolean secure;
+
+ private final int readTimeoutMs;
+
+ private String clientId;
+
+ private NioEventLoopGroup group;
+
+ private Channel channel;
+
+ StandbyClient(String clientId, boolean secure, int readTimeoutMs) {
+ this.clientId = clientId;
+ this.secure = secure;
+ this.readTimeoutMs = readTimeoutMs;
+ }
+
+ void connect(String host, int port) throws Exception {
+ group = new NioEventLoopGroup();
+
+ final SslContext sslContext;
+
+ if (secure) {
+ sslContext =
SslContext.newClientContext(InsecureTrustManagerFactory.INSTANCE);
+ } else {
+ sslContext = null;
+ }
+
+ Bootstrap b = new Bootstrap()
+ .group(group)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, readTimeoutMs)
+ .option(ChannelOption.TCP_NODELAY, true)
+ .option(ChannelOption.SO_REUSEADDR, true)
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .handler(new ChannelInitializer<SocketChannel>() {
+
+ @Override
+ public void initChannel(SocketChannel ch) throws Exception
{
+ ChannelPipeline p = ch.pipeline();
+
+ if (sslContext != null) {
+ p.addLast(sslContext.newHandler(ch.alloc()));
+ }
+
+ p.addLast(new ReadTimeoutHandler(readTimeoutMs,
TimeUnit.MILLISECONDS));
+
+ // Decoders
+
+ p.addLast(new SnappyFramedDecoder(true));
+
+ // Such a big max frame length is needed because blob
+ // values are sent in one big message. In future
+ // versions of the protocol, sending binaries in chunks
+ // should be considered instead.
+
+ p.addLast(new
LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4));
+ p.addLast(new ResponseDecoder());
+
+ // Encoders
+
+ p.addLast(new StringEncoder(CharsetUtil.UTF_8));
+ p.addLast(new GetHeadRequestEncoder());
+ p.addLast(new GetSegmentRequestEncoder());
+ p.addLast(new GetBlobRequestEncoder());
+
+ // Handlers
+
+ p.addLast(new GetHeadResponseHandler(headQueue));
+ p.addLast(new GetSegmentResponseHandler(segmentQueue));
+ p.addLast(new GetBlobResponseHandler(blobQueue));
+ }
+
+ });
+
+ channel = b.connect(host, port).sync().channel();
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ channel.close().sync();
+ channel = null;
+
+ group.shutdownGracefully();
+ group = null;
+ }
+
+ String getHead() throws InterruptedException {
+ channel.writeAndFlush(new GetHeadRequest(clientId));
+
+ GetHeadResponse response = headQueue.poll(readTimeoutMs,
TimeUnit.MILLISECONDS);
+
+ if (response == null) {
+ return null;
+ }
+
+ return response.getHeadRecordId();
+ }
+
+ byte[] getSegment(String segmentId) throws InterruptedException {
+ channel.writeAndFlush(new GetSegmentRequest(clientId, segmentId));
+
+ GetSegmentResponse response = segmentQueue.poll(readTimeoutMs,
TimeUnit.MILLISECONDS);
+
+ if (response == null) {
+ return null;
+ }
+
+ return response.getSegmentData();
+ }
+
+ byte[] getBlob(String blobId) throws InterruptedException {
+ channel.writeAndFlush(new GetBlobRequest(clientId, blobId));
+
+ GetBlobResponse response = blobQueue.poll(readTimeoutMs,
TimeUnit.MILLISECONDS);
+
+ if (response == null) {
+ return null;
+ }
+
+ return response.getBlobData();
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
------------------------------------------------------------------------------
svn:eol-style = native
Copied:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyDiff.java
(from r1760420,
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyApplyDiff.java)
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyDiff.java?p2=jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyDiff.java&p1=jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyApplyDiff.java&r1=1760420&r2=1760934&rev=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyApplyDiff.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyDiff.java
Thu Sep 15 14:03:05 2016
@@ -22,8 +22,10 @@ package org.apache.jackrabbit.oak.segmen
import static org.apache.jackrabbit.oak.api.Type.BINARIES;
import static org.apache.jackrabbit.oak.api.Type.BINARY;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
+import com.google.common.base.Supplier;
import org.apache.jackrabbit.oak.api.Blob;
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.api.Type;
@@ -31,29 +33,29 @@ import org.apache.jackrabbit.oak.plugins
import org.apache.jackrabbit.oak.segment.RecordId;
import org.apache.jackrabbit.oak.segment.SegmentBlob;
import org.apache.jackrabbit.oak.segment.SegmentNodeState;
-import org.apache.jackrabbit.oak.segment.standby.store.RemoteSegmentLoader;
-import org.apache.jackrabbit.oak.segment.standby.store.StandbyStore;
+import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.apache.jackrabbit.oak.spi.state.NodeStateDiff;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-class StandbyApplyDiff implements NodeStateDiff {
+class StandbyDiff implements NodeStateDiff {
- private static final Logger log = LoggerFactory
- .getLogger(StandbyApplyDiff.class);
+ private static final Logger log =
LoggerFactory.getLogger(StandbyDiff.class);
private final NodeBuilder builder;
- private final StandbyStore store;
+ private final FileStore store;
- private final boolean hasDataStore;
+ private final StandbyClient client;
- private final RemoteSegmentLoader loader;
+ private final boolean hasDataStore;
private final String path;
+ private final Supplier<Boolean> running;
+
/**
* read-only traversal of the diff that has 2 properties: one is to log all
* the content changes, second is to drill down to properly level, so that
@@ -61,60 +63,70 @@ class StandbyApplyDiff implements NodeSt
*/
private final boolean logOnly;
- public StandbyApplyDiff(NodeBuilder builder, StandbyStore store,
- RemoteSegmentLoader loader) {
- this(builder, store, loader, "/", false);
+ StandbyDiff(NodeBuilder builder, FileStore store, StandbyClient client,
Supplier<Boolean> running) {
+ this(builder, store, client, "/", false, running);
}
- private StandbyApplyDiff(NodeBuilder builder, StandbyStore store,
- RemoteSegmentLoader loader, String path, boolean logOnly) {
+ private StandbyDiff(NodeBuilder builder, FileStore store, StandbyClient
client, String path, boolean logOnly, Supplier<Boolean> running) {
this.builder = builder;
this.store = store;
this.hasDataStore = store.getBlobStore() != null;
- this.loader = loader;
+ this.client = client;
this.path = path;
this.logOnly = logOnly;
+ this.running = running;
+ }
+
+ private boolean stop() {
+ return !running.get();
}
@Override
public boolean propertyAdded(PropertyState after) {
- if (!loader.isRunning()) {
+ if (stop()) {
return false;
}
- if (!logOnly) {
- builder.setProperty(binaryCheck(after));
- } else {
+
+ if (logOnly) {
binaryCheck(after);
+ } else {
+ builder.setProperty(binaryCheck(after));
}
+
return true;
}
@Override
public boolean propertyChanged(PropertyState before, PropertyState after) {
- if (!loader.isRunning()) {
+ if (stop()) {
return false;
}
- if (!logOnly) {
- builder.setProperty(binaryCheck(after));
- } else {
+
+ if (logOnly) {
binaryCheck(after);
+ } else {
+ builder.setProperty(binaryCheck(after));
}
+
return true;
}
@Override
public boolean propertyDeleted(PropertyState before) {
- if (!loader.isRunning()) {
+ if (stop()) {
return false;
}
+
if (!logOnly) {
builder.removeProperty(before.getName());
}
+
return true;
}
private PropertyState binaryCheck(PropertyState property) {
Type<?> type = property.getType();
+
if (type == BINARY) {
binaryCheck(property.getValue(Type.BINARY), property.getName());
} else if (type == BINARIES) {
@@ -122,93 +134,100 @@ class StandbyApplyDiff implements NodeSt
binaryCheck(blob, property.getName());
}
}
+
return property;
}
private void binaryCheck(Blob b, String pName) {
if (b instanceof SegmentBlob) {
- SegmentBlob sb = (SegmentBlob) b;
- // verify if the blob exists
- if (sb.isExternal() && hasDataStore && b.getReference() == null) {
- String blobId = sb.getBlobId();
- if (blobId != null) {
- readBlob(blobId, pName);
- }
- }
+ binaryCheck((SegmentBlob) b, pName);
} else {
- log.warn("Unknown Blob {} at {}, ignoring", b.getClass().getName(),
- path + "#" + pName);
+ log.warn("Unknown Blob {} at {}, ignoring",
b.getClass().getName(), path + "#" + pName);
}
}
- private void readBlob(String blobId, String pName) {
- Blob read = loader.readBlob(blobId);
- if (read != null) {
+ private void binaryCheck(SegmentBlob sb, String pName) {
+ if (sb.isExternal() && hasDataStore && sb.getReference() == null) {
+ String blobId = sb.getBlobId();
+
+ if (blobId == null) {
+ return;
+ }
+
try {
- store.getBlobStore().writeBlob(read.getNewStream());
- } catch (IOException f) {
- throw new IllegalStateException("Unable to persist blob "
- + blobId + " at " + path + "#" + pName, f);
+ readBlob(blobId, pName);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
}
- } else {
- throw new IllegalStateException("Unable to load remote blob "
- + blobId + " at " + path + "#" + pName);
+ }
+ }
+
+ private void readBlob(String blobId, String pName) throws
InterruptedException {
+ byte[] data = client.getBlob(blobId);
+
+ if (data == null) {
+ throw new IllegalStateException("Unable to load remote blob " +
blobId + " at " + path + "#" + pName);
+ }
+
+ try {
+ store.getBlobStore().writeBlob(new ByteArrayInputStream(data));
+ } catch (IOException f) {
+ throw new IllegalStateException("Unable to persist blob " + blobId
+ " at " + path + "#" + pName, f);
}
}
@Override
public boolean childNodeAdded(String name, NodeState after) {
- return process(name, "childNodeAdded", EmptyNodeState.EMPTY_NODE,
- after);
+ return process(name, "childNodeAdded", EmptyNodeState.EMPTY_NODE,
after);
}
@Override
- public boolean childNodeChanged(String name, NodeState before,
- NodeState after) {
+ public boolean childNodeChanged(String name, NodeState before, NodeState
after) {
return process(name, "childNodeChanged", before, after);
}
- private boolean process(String name, String op, NodeState before,
- NodeState after) {
- if (!loader.isRunning()) {
+ private boolean process(String name, String op, NodeState before,
NodeState after) {
+ if (stop()) {
return false;
}
+
if (after instanceof SegmentNodeState) {
if (log.isTraceEnabled()) {
- log.trace("{} {}, readonly binary check {}", op, path + name,
- logOnly);
+ log.trace("{} {}, readonly binary check {}", op, path + name,
logOnly);
}
+
if (!logOnly) {
RecordId id = ((SegmentNodeState) after).getRecordId();
- builder.setChildNode(name, store.newSegmentNodeState(id));
+ builder.setChildNode(name, store.getReader().readNode(id));
}
+
if ("checkpoints".equals(name)) {
// if we're on the /checkpoints path, there's no need for a
deep
// traversal to verify binaries
return true;
}
- if (hasDataStore) {
- // has external datastore, we need a deep
- // traversal to verify binaries
- return after.compareAgainstBaseState(before,
- new StandbyApplyDiff(builder.getChildNode(name), store,
- loader, path + name + "/", true));
- } else {
+
+ if (!hasDataStore) {
return true;
+
}
+ // has external datastore, we need a deep
+ // traversal to verify binaries
+ return after.compareAgainstBaseState(before, new
StandbyDiff(builder.getChildNode(name), store, client, path + name + "/", true,
running));
}
+
return false;
}
@Override
public boolean childNodeDeleted(String name, NodeState before) {
- if (!loader.isRunning()) {
- return false;
- }
log.trace("childNodeDeleted {}, RO:{}", path + name, logOnly);
+
if (!logOnly) {
builder.getChildNode(name).remove();
}
+
return true;
}
+
}
Copied:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySync.java
(from r1760651,
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java)
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySync.java?p2=jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySync.java&p1=jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java&r1=1760651&r2=1760934&rev=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbyClient.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySync.java
Thu Sep 15 14:03:05 2016
@@ -16,12 +16,15 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.jackrabbit.oak.segment.standby.client;
+import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;
+
import java.io.Closeable;
+import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.UUID;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.management.MBeanServer;
@@ -29,61 +32,51 @@ import javax.management.ObjectName;
import javax.management.StandardMBean;
import javax.net.ssl.SSLException;
-import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.handler.codec.compression.SnappyFramedDecoder;
-import io.netty.handler.codec.string.StringEncoder;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import io.netty.handler.timeout.ReadTimeoutHandler;
-import io.netty.util.CharsetUtil;
-import org.apache.jackrabbit.oak.segment.SegmentStore;
+import com.google.common.base.Supplier;
import org.apache.jackrabbit.oak.segment.file.FileStore;
-import org.apache.jackrabbit.oak.segment.standby.codec.RecordIdDecoder;
import org.apache.jackrabbit.oak.segment.standby.jmx.ClientStandbyStatusMBean;
import org.apache.jackrabbit.oak.segment.standby.jmx.StandbyStatusMBean;
import org.apache.jackrabbit.oak.segment.standby.store.CommunicationObserver;
-import org.apache.jackrabbit.oak.segment.standby.store.StandbyStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public final class StandbyClient implements ClientStandbyStatusMBean,
Runnable, Closeable {
+public final class StandbySync implements ClientStandbyStatusMBean, Runnable,
Closeable {
+
public static final String CLIENT_ID_PROPERTY_NAME = "standbyID";
- private static final Logger log = LoggerFactory
- .getLogger(StandbyClient.class);
+ private static final Logger log =
LoggerFactory.getLogger(StandbySync.class);
private final String host;
+
private final int port;
+
private final int readTimeoutMs;
+
private final boolean autoClean;
- private final StandbyStore store;
private final CommunicationObserver observer;
- private StandbyClientHandler handler;
- private EventLoopGroup group;
- private SslContext sslContext;
+
+ private final boolean secure;
+
private boolean active = false;
+
private int failedRequests;
+
private long lastSuccessfulRequest;
+
private volatile String state;
+
private final Object sync = new Object();
+ private final FileStore fileStore;
+
private final AtomicBoolean running = new AtomicBoolean(true);
private long syncStartTimestamp;
+
private long syncEndTimestamp;
- public StandbyClient(String host, int port, FileStore store,
- boolean secure, int readTimeoutMs, boolean autoClean)
- throws SSLException {
+ public StandbySync(String host, int port, FileStore store, boolean secure,
int readTimeoutMs, boolean autoClean) throws SSLException {
this.state = STATUS_INITIALIZING;
this.lastSuccessfulRequest = -1;
this.syncStartTimestamp = -1;
@@ -91,20 +84,17 @@ public final class StandbyClient impleme
this.failedRequests = 0;
this.host = host;
this.port = port;
- if (secure) {
- this.sslContext =
SslContext.newClientContext(InsecureTrustManagerFactory.INSTANCE);
- }
+ this.secure = secure;
this.readTimeoutMs = readTimeoutMs;
this.autoClean = autoClean;
- this.store = new StandbyStore(store);
+ this.fileStore = store;
String s = System.getProperty(CLIENT_ID_PROPERTY_NAME);
this.observer = new CommunicationObserver((s == null || s.length() ==
0) ? UUID.randomUUID().toString() : s);
final MBeanServer jmxServer =
ManagementFactory.getPlatformMBeanServer();
try {
jmxServer.registerMBean(new StandardMBean(this,
ClientStandbyStatusMBean.class), new ObjectName(this.getMBeanName()));
- }
- catch (Exception e) {
+ } catch (Exception e) {
log.error("can register standby status mbean", e);
}
}
@@ -119,12 +109,10 @@ public final class StandbyClient impleme
final MBeanServer jmxServer =
ManagementFactory.getPlatformMBeanServer();
try {
jmxServer.unregisterMBean(new ObjectName(this.getMBeanName()));
- }
- catch (Exception e) {
+ } catch (Exception e) {
log.error("can unregister standby status mbean", e);
}
observer.unregister();
- shutdownNetty();
state = STATUS_CLOSED;
}
@@ -134,49 +122,33 @@ public final class StandbyClient impleme
return;
}
- Bootstrap b;
- synchronized (this.sync) {
- if (this.active) {
+ state = STATUS_STARTING;
+
+ synchronized (sync) {
+ if (active) {
return;
}
- state = STATUS_STARTING;
- handler = new StandbyClientHandler(this.store, observer, running,
- readTimeoutMs, autoClean);
- group = new NioEventLoopGroup();
-
- b = new Bootstrap();
- b.group(group);
- b.channel(NioSocketChannel.class);
- b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, readTimeoutMs);
- b.option(ChannelOption.TCP_NODELAY, true);
- b.option(ChannelOption.SO_REUSEADDR, true);
- b.option(ChannelOption.SO_KEEPALIVE, true);
-
- b.handler(new ChannelInitializer<SocketChannel>() {
- @Override
- public void initChannel(SocketChannel ch) throws Exception {
- ChannelPipeline p = ch.pipeline();
- if (sslContext != null) {
- p.addLast(sslContext.newHandler(ch.alloc()));
- }
- p.addLast("readTimeoutHandler", new ReadTimeoutHandler(
- readTimeoutMs, TimeUnit.MILLISECONDS));
- p.addLast(new StringEncoder(CharsetUtil.UTF_8));
- p.addLast(new SnappyFramedDecoder(true));
- p.addLast(new RecordIdDecoder(store));
- p.addLast(handler);
- }
- });
state = STATUS_RUNNING;
- this.active = true;
+ active = true;
}
try {
long startTimestamp = System.currentTimeMillis();
- // Start the client.
- ChannelFuture f = b.connect(host, port).sync();
- // Wait until the connection is closed.
- f.channel().closeFuture().sync();
+ try (StandbyClient client = new StandbyClient(observer.getID(),
secure, readTimeoutMs)) {
+ client.connect(host, port);
+
+ long sizeBefore = fileStore.getStats().getApproximateSize();
+ new StandbySyncExecution(fileStore, client,
newRunningSupplier()).execute();
+ long sizeAfter = fileStore.getStats().getApproximateSize();
+
+ if (autoClean && sizeBefore > 0) {
+ // if size gain is over 25% call cleanup
+ if (sizeAfter - sizeBefore > 0.25 * sizeBefore) {
+ log.info("Store size increased from {} to {}, will run
cleanup.", humanReadableByteCount(sizeBefore),
humanReadableByteCount(sizeAfter));
+ fileStore.cleanup();
+ }
+ }
+ }
this.failedRequests = 0;
this.syncStartTimestamp = startTimestamp;
this.syncEndTimestamp = System.currentTimeMillis();
@@ -187,20 +159,19 @@ public final class StandbyClient impleme
} finally {
synchronized (this.sync) {
this.active = false;
- shutdownNetty();
}
}
}
- private void shutdownNetty() {
- if (handler != null) {
- handler.close();
- handler = null;
- }
- if (group != null && !group.isShuttingDown()) {
- group.shutdownGracefully(0, 1,
TimeUnit.SECONDS).syncUninterruptibly();
- group = null;
- }
+ private Supplier<Boolean> newRunningSupplier() {
+ return new Supplier<Boolean>() {
+
+ @Override
+ public Boolean get() {
+ return running.get();
+ }
+
+ };
}
@Override
@@ -237,8 +208,10 @@ public final class StandbyClient impleme
@Override
public int getSecondsSinceLastSuccess() {
- if (this.lastSuccessfulRequest < 0) return -1;
- return (int)(System.currentTimeMillis() / 1000 -
this.lastSuccessfulRequest);
+ if (this.lastSuccessfulRequest < 0) {
+ return -1;
+ }
+ return (int) (System.currentTimeMillis() / 1000 -
this.lastSuccessfulRequest);
}
@Override
@@ -253,7 +226,11 @@ public final class StandbyClient impleme
@Override
public void cleanup() {
- store.cleanup();
+ try {
+ fileStore.cleanup();
+ } catch (IOException e) {
+ log.error("Error while cleaning up", e);
+ }
}
@Override
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySyncExecution.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySyncExecution.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySyncExecution.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySyncExecution.java
Thu Sep 15 14:03:05 2016
@@ -0,0 +1,166 @@
+/*
+ * 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.client;
+
+import static com.google.common.collect.Maps.newHashMap;
+import static com.google.common.collect.Sets.newHashSet;
+
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+
+import javax.annotation.Nonnull;
+
+import com.google.common.base.Supplier;
+import org.apache.jackrabbit.oak.segment.RecordId;
+import org.apache.jackrabbit.oak.segment.Segment;
+import org.apache.jackrabbit.oak.segment.SegmentId;
+import org.apache.jackrabbit.oak.segment.SegmentNodeBuilder;
+import org.apache.jackrabbit.oak.segment.SegmentNodeState;
+import org.apache.jackrabbit.oak.segment.SegmentNotFoundException;
+import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Encapsulates the algorithm for a single execution of the synchronization
+ * process between the primary and the standby instance. It also contains
+ * temporary state that is supposed to be used for the lifetime of a
+ * synchronization run.
+ */
+class StandbySyncExecution {
+
+ private static final Logger log =
LoggerFactory.getLogger(StandbySyncExecution.class);
+
+ private final FileStore store;
+
+ private final StandbyClient client;
+
+ private final Supplier<Boolean> running;
+
+ private final Set<UUID> visited = newHashSet();
+
+ private final Map<UUID, Segment> cache = newHashMap();
+
+ StandbySyncExecution(FileStore store, StandbyClient client,
Supplier<Boolean> running) {
+ this.store = store;
+ this.client = client;
+ this.running = running;
+ }
+
+ void execute() throws Exception {
+ RecordId remoteHead = getHead();
+
+ if (remoteHead.equals(store.getHead().getRecordId())) {
+ return;
+ }
+
+ long t = System.currentTimeMillis();
+ SegmentNodeState before = store.getHead();
+ SegmentNodeBuilder builder = before.builder();
+ SegmentNodeState current = newSegmentNodeState(remoteHead);
+ compareAgainstBaseState(current, before, builder);
+ boolean ok = setHead(before, builder.getNodeState());
+ log.debug("updated head state successfully: {} in {}ms.", ok,
System.currentTimeMillis() - t);
+ }
+
+ private RecordId getHead() throws Exception {
+ return RecordId.fromString(store, client.getHead());
+ }
+
+ private SegmentNodeState newSegmentNodeState(RecordId id) {
+ return store.getReader().readNode(id);
+ }
+
+ private boolean setHead(@Nonnull SegmentNodeState expected, @Nonnull
SegmentNodeState head) {
+ return store.getRevisions().setHead(expected.getRecordId(),
head.getRecordId());
+ }
+
+ private boolean compareAgainstBaseState(SegmentNodeState current,
SegmentNodeState before, SegmentNodeBuilder builder) throws Exception {
+ while (true) {
+ try {
+ return current.compareAgainstBaseState(before, new
StandbyDiff(builder, store, client, running));
+ } catch (SegmentNotFoundException e) {
+ log.info("Found missing segment {}", e.getSegmentId());
+
copySegmentHierarchyFromPrimary(UUID.fromString(e.getSegmentId()));
+ }
+ }
+ }
+
+ private void copySegmentHierarchyFromPrimary(UUID segmentId) throws
Exception {
+ LinkedList<UUID> batch = new LinkedList<>();
+
+ batch.offer(segmentId);
+
+ while (batch.size() > 0) {
+ UUID current = batch.remove();
+
+ log.info("Loading segment {}", current);
+ Segment segment = copySegmentFromPrimary(current);
+
+ log.info("Marking segment {} as loaded", current);
+ visited.add(current);
+
+ if (!SegmentId.isDataSegmentId(current.getLeastSignificantBits()))
{
+ continue;
+ }
+
+ log.info("Inspecting segment {} for references", current);
+ for (int i = 0; i < segment.getReferencedSegmentIdCount(); i++) {
+ UUID referenced = segment.getReferencedSegmentId(i);
+
+ if (visited.contains(referenced)) {
+ continue;
+ }
+
+ log.info("Found reference from {} to {}", current, referenced);
+ if
(SegmentId.isDataSegmentId(referenced.getLeastSignificantBits())) {
+ batch.add(referenced);
+ } else {
+ batch.addFirst(referenced);
+ }
+ }
+ }
+ }
+
+ private Segment copySegmentFromPrimary(UUID uuid) throws Exception {
+ Segment result = cache.get(uuid);
+
+ if (result != null) {
+ log.info("Segment {} was found in the local cache", uuid);
+ return result;
+ }
+
+ byte[] data = client.getSegment(uuid.toString());
+
+ if (data == null) {
+ throw new IllegalStateException("Unable to read segment " + uuid);
+ }
+
+ long msb = uuid.getMostSignificantBits();
+ long lsb = uuid.getLeastSignificantBits();
+ SegmentId segmentId = store.newSegmentId(msb, lsb);
+ store.writeSegment(segmentId, data, 0, data.length);
+ result = segmentId.getSegment();
+ cache.put(uuid, result);
+ return result;
+ }
+
+}
+
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/client/StandbySyncExecution.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestEncoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestEncoder.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestEncoder.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestEncoder.java
Thu Sep 15 14:03:05 2016
@@ -0,0 +1,37 @@
+/*
+ * 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.codec;
+
+import java.util.List;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GetBlobRequestEncoder extends
MessageToMessageEncoder<GetBlobRequest> {
+
+ private final Logger log =
LoggerFactory.getLogger(GetBlobRequestEncoder.class);
+
+ @Override
+ protected void encode(ChannelHandlerContext ctx, GetBlobRequest msg,
List<Object> out) throws Exception {
+ log.debug("Sending request from client {} for blob {}",
msg.getClientId(), msg.getBlobId());
+ out.add(Messages.newGetBlobRequest(msg.getClientId(),
msg.getBlobId()));
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobRequestEncoder.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponse.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponse.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponse.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponse.java
Thu Sep 15 14:03:05 2016
@@ -17,25 +17,30 @@
package org.apache.jackrabbit.oak.segment.standby.codec;
-import org.apache.jackrabbit.oak.api.Blob;
-
public class GetBlobResponse {
private final String clientId;
- private final Blob blob;
+ private final String blobId;
+
+ private final byte[] blobData;
- public GetBlobResponse(String clientId, Blob blob) {
+ public GetBlobResponse(String clientId, String blobId, byte[] blobData) {
this.clientId = clientId;
- this.blob = blob;
+ this.blobId = blobId;
+ this.blobData = blobData;
}
public String getClientId() {
return clientId;
}
- public Blob getBlob() {
- return blob;
+ public String getBlobId() {
+ return blobId;
+ }
+
+ public byte[] getBlobData() {
+ return blobData;
}
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponseEncoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponseEncoder.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponseEncoder.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetBlobResponseEncoder.java
Thu Sep 15 14:03:05 2016
@@ -17,7 +17,6 @@
package org.apache.jackrabbit.oak.segment.standby.codec;
-import java.io.InputStream;
import java.nio.charset.Charset;
import com.google.common.hash.Hasher;
@@ -25,8 +24,6 @@ import com.google.common.hash.Hashing;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
-import org.apache.commons.io.IOUtils;
-import org.apache.jackrabbit.oak.api.Blob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,30 +33,22 @@ public class GetBlobResponseEncoder exte
@Override
protected void encode(ChannelHandlerContext ctx, GetBlobResponse msg,
ByteBuf out) throws Exception {
- log.debug("Sending blob {} to client {}",
msg.getBlob().getContentIdentity(), msg.getClientId());
- encode(msg.getBlob(), out);
+ log.debug("Sending blob {} to client {}", msg.getBlobId(),
msg.getClientId());
+ encode(msg.getBlobId(), msg.getBlobData(), out);
}
- private void encode(Blob b, ByteBuf out) throws Exception {
- byte[] bytes;
-
- try (InputStream s = b.getNewStream()) {
- bytes = IOUtils.toByteArray(s);
- }
+ private void encode(String blobId, byte[] data, ByteBuf out) throws
Exception {
+ byte[] blobIdBytes = blobId.getBytes(Charset.forName("UTF-8"));
Hasher hasher = Hashing.murmur3_32().newHasher();
- long hash = hasher.putBytes(bytes).hash().padToLong();
+ long hash = hasher.putBytes(data).hash().padToLong();
- out.writeInt(bytes.length);
+ out.writeInt(1 + 4 + blobIdBytes.length + 8 + data.length);
out.writeByte(Messages.HEADER_BLOB);
-
- String bid = b.getContentIdentity();
- byte[] id = bid.getBytes(Charset.forName("UTF-8"));
- out.writeInt(id.length);
- out.writeBytes(id);
-
+ out.writeInt(blobIdBytes.length);
+ out.writeBytes(blobIdBytes);
out.writeLong(hash);
- out.writeBytes(bytes);
+ out.writeBytes(data);
}
}
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestEncoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestEncoder.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestEncoder.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestEncoder.java
Thu Sep 15 14:03:05 2016
@@ -0,0 +1,37 @@
+/*
+ * 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.codec;
+
+import java.util.List;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GetHeadRequestEncoder extends
MessageToMessageEncoder<GetHeadRequest> {
+
+ private static final Logger log =
LoggerFactory.getLogger(GetHeadRequestEncoder.class);
+
+ @Override
+ protected void encode(ChannelHandlerContext ctx, GetHeadRequest msg,
List<Object> out) throws Exception {
+ log.debug("Sending request from client {} for current head",
msg.getClientId());
+ out.add(Messages.newGetHeadRequest(msg.getClientId()));
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadRequestEncoder.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponse.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponse.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponse.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponse.java
Thu Sep 15 14:03:05 2016
@@ -17,15 +17,13 @@
package org.apache.jackrabbit.oak.segment.standby.codec;
-import org.apache.jackrabbit.oak.segment.RecordId;
-
public class GetHeadResponse {
private final String clientId;
- private final RecordId headRecordId;
+ private final String headRecordId;
- public GetHeadResponse(String clientId, RecordId headRecordId) {
+ public GetHeadResponse(String clientId, String headRecordId) {
this.clientId = clientId;
this.headRecordId = headRecordId;
}
@@ -34,7 +32,7 @@ public class GetHeadResponse {
return clientId;
}
- public RecordId getHeadRecordId() {
+ public String getHeadRecordId() {
return headRecordId;
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponseEncoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponseEncoder.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponseEncoder.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetHeadResponseEncoder.java
Thu Sep 15 14:03:05 2016
@@ -34,7 +34,7 @@ public class GetHeadResponseEncoder exte
@Override
protected void encode(ChannelHandlerContext ctx, GetHeadResponse msg,
ByteBuf out) throws Exception {
log.debug("Sending head {} to client {}", msg.getHeadRecordId(),
msg.getClientId());
- byte[] body =
msg.getHeadRecordId().toString().getBytes(CharsetUtil.UTF_8);
+ byte[] body = msg.getHeadRecordId().getBytes(CharsetUtil.UTF_8);
out.writeInt(body.length + 1);
out.writeByte(Messages.HEADER_RECORD);
out.writeBytes(body);
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequest.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequest.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequest.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequest.java
Thu Sep 15 14:03:05 2016
@@ -17,15 +17,13 @@
package org.apache.jackrabbit.oak.segment.standby.codec;
-import java.util.UUID;
-
public class GetSegmentRequest {
private final String clientId;
- private final UUID segmentId;
+ private final String segmentId;
- public GetSegmentRequest(String clientId, UUID segmentId) {
+ public GetSegmentRequest(String clientId, String segmentId) {
this.clientId = clientId;
this.segmentId = segmentId;
}
@@ -34,7 +32,7 @@ public class GetSegmentRequest {
return clientId;
}
- public UUID getSegmentId() {
+ public String getSegmentId() {
return segmentId;
}
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestEncoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestEncoder.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestEncoder.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestEncoder.java
Thu Sep 15 14:03:05 2016
@@ -0,0 +1,37 @@
+/*
+ * 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.codec;
+
+import java.util.List;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GetSegmentRequestEncoder extends
MessageToMessageEncoder<GetSegmentRequest> {
+
+ private final Logger log =
LoggerFactory.getLogger(GetSegmentRequestEncoder.class);
+
+ @Override
+ protected void encode(ChannelHandlerContext ctx, GetSegmentRequest msg,
List<Object> out) throws Exception {
+ log.debug("Sending request from client {} for segment {}",
msg.getClientId(), msg.getSegmentId());
+ out.add(Messages.newGetSegmentRequest(msg.getClientId(),
msg.getSegmentId()));
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentRequestEncoder.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponse.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponse.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponse.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponse.java
Thu Sep 15 14:03:05 2016
@@ -17,25 +17,30 @@
package org.apache.jackrabbit.oak.segment.standby.codec;
-import org.apache.jackrabbit.oak.segment.Segment;
-
public class GetSegmentResponse {
private final String clientId;
- private final Segment segment;
+ private final String segmentId;
+
+ private final byte[] segmentData;
- public GetSegmentResponse(String clientId, Segment segment) {
+ public GetSegmentResponse(String clientId, String segmentId, byte[]
segmentData) {
this.clientId = clientId;
- this.segment = segment;
+ this.segmentId = segmentId;
+ this.segmentData = segmentData;
}
public String getClientId() {
return clientId;
}
- public Segment getSegment() {
- return segment;
+ public String getSegmentId() {
+ return segmentId;
+ }
+
+ public byte[] getSegmentData() {
+ return segmentData;
}
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponseEncoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponseEncoder.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponseEncoder.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/GetSegmentResponseEncoder.java
Thu Sep 15 14:03:05 2016
@@ -17,15 +17,13 @@
package org.apache.jackrabbit.oak.segment.standby.codec;
-import java.io.ByteArrayOutputStream;
+import java.util.UUID;
import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
-import org.apache.jackrabbit.oak.segment.Segment;
-import org.apache.jackrabbit.oak.segment.SegmentId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,27 +40,23 @@ public class GetSegmentResponseEncoder e
@Override
protected void encode(ChannelHandlerContext ctx, GetSegmentResponse msg,
ByteBuf out) throws Exception {
- log.debug("Sending segment {} to client {}",
msg.getSegment().getSegmentId(), msg.getClientId());
- encode(msg.getSegment(), out);
+ log.debug("Sending segment {} to client {}", msg.getSegmentId(),
msg.getClientId());
+ encode(msg.getSegmentId(), msg.getSegmentData(), out);
}
- private void encode(Segment s, ByteBuf out) throws Exception {
- SegmentId id = s.getSegmentId();
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream(s.size());
- s.writeTo(baos);
- byte[] segment = baos.toByteArray();
+ private void encode(String segmentId, byte[] data, ByteBuf out) throws
Exception {
+ UUID id = UUID.fromString(segmentId);
Hasher hasher = Hashing.murmur3_32().newHasher();
- long hash = hasher.putBytes(segment).hash().padToLong();
+ long hash = hasher.putBytes(data).hash().padToLong();
- int len = segment.length + EXTRA_HEADERS_WO_SIZE;
+ int len = data.length + EXTRA_HEADERS_WO_SIZE;
out.writeInt(len);
out.writeByte(Messages.HEADER_SEGMENT);
out.writeLong(id.getMostSignificantBits());
out.writeLong(id.getLeastSignificantBits());
out.writeLong(hash);
- out.writeBytes(segment);
+ out.writeBytes(data);
}
}
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=1760934&r1=1760933&r2=1760934&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
Thu Sep 15 14:03:05 2016
@@ -16,19 +16,25 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.jackrabbit.oak.segment.standby.codec;
-public class Messages {
+class Messages {
+
+ static final byte HEADER_RECORD = 0x00;
+
+ static final byte HEADER_SEGMENT = 0x01;
+
+ static final byte HEADER_BLOB = 0x02;
- public static final byte HEADER_RECORD = 0x00;
- public static final byte HEADER_SEGMENT = 0x01;
- public static final byte HEADER_BLOB = 0x02;
-
- public static final String GET_HEAD = "h";
- public static final String GET_SEGMENT = "s.";
- public static final String GET_BLOB = "b.";
+ static final String GET_HEAD = "h";
+
+ static final String GET_SEGMENT = "s.";
+
+ static final String GET_BLOB = "b.";
private static final String MAGIC = "Standby-CMD@";
+
private static final String SEPARATOR = ":";
private static String newRequest(String clientId, String body, boolean
delimited) {
@@ -48,31 +54,31 @@ public class Messages {
return builder.toString();
}
- public static String newGetHeadRequest(String clientId, boolean delimited)
{
+ static String newGetHeadRequest(String clientId, boolean delimited) {
return newRequest(clientId, GET_HEAD, delimited);
}
- public static String newGetHeadRequest(String clientId) {
+ static String newGetHeadRequest(String clientId) {
return newGetHeadRequest(clientId, true);
}
- public static String newGetSegmentRequest(String clientId, String
segmentId, boolean delimited) {
+ static String newGetSegmentRequest(String clientId, String segmentId,
boolean delimited) {
return newRequest(clientId, GET_SEGMENT + segmentId, delimited);
}
- public static String newGetSegmentRequest(String clientId, String
segmentId) {
+ static String newGetSegmentRequest(String clientId, String segmentId) {
return newGetSegmentRequest(clientId, segmentId, true);
}
- public static String newGetBlobRequest(String clientId, String blobId,
boolean delimited) {
+ static String newGetBlobRequest(String clientId, String blobId, boolean
delimited) {
return newRequest(clientId, GET_BLOB + blobId, delimited);
}
- public static String newGetBlobRequest(String clientId, String blobId) {
+ static String newGetBlobRequest(String clientId, String blobId) {
return newGetBlobRequest(clientId, blobId, true);
}
- public static String extractMessageFrom(String payload) {
+ static String extractMessageFrom(String payload) {
if (payload.startsWith(MAGIC) && payload.length() > MAGIC.length()) {
int i = payload.indexOf(SEPARATOR);
return payload.substring(i + 1);
@@ -80,7 +86,7 @@ public class Messages {
return null;
}
- public static String extractClientFrom(String payload) {
+ static String extractClientFrom(String payload) {
if (payload.startsWith(MAGIC) && payload.length() > MAGIC.length()) {
payload = payload.substring(MAGIC.length());
int i = payload.indexOf(SEPARATOR);
@@ -88,4 +94,5 @@ public class Messages {
}
return null;
}
+
}
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RequestDecoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RequestDecoder.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RequestDecoder.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RequestDecoder.java
Thu Sep 15 14:03:05 2016
@@ -0,0 +1,52 @@
+/*
+ * 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.codec;
+
+import java.util.List;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public 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("Received invalid message {}, ignoring", msg);
+ ctx.fireChannelRead(msg);
+ } else if (request.startsWith(Messages.GET_BLOB)) {
+ log.debug("Parsed 'get blob' request");
+ out.add(new GetBlobRequest(Messages.extractClientFrom(msg),
request.substring(Messages.GET_BLOB.length())));
+ } 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 {
+ log.debug("Received unrecognizable message {}, dropping", msg);
+ }
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/RequestDecoder.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/ResponseDecoder.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/ResponseDecoder.java?rev=1760934&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/ResponseDecoder.java
(added)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/ResponseDecoder.java
Thu Sep 15 14:03:05 2016
@@ -0,0 +1,108 @@
+/*
+ * 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.codec;
+
+import java.util.List;
+import java.util.UUID;
+
+import com.google.common.base.Charsets;
+import com.google.common.hash.Hashing;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ResponseDecoder extends ByteToMessageDecoder {
+
+ private static final Logger log =
LoggerFactory.getLogger(ResponseDecoder.class);
+
+ @Override
+ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object>
out) throws Exception {
+ int length = in.readInt();
+
+ switch (in.readByte()) {
+ case Messages.HEADER_RECORD:
+ log.debug("Decoding 'get head' response");
+ decodeGetHeadResponse(length, in, out);
+ break;
+ case Messages.HEADER_SEGMENT:
+ log.debug("Decoding 'get segment' response");
+ decodeGetSegmentResponse(length, in, out);
+ break;
+ case Messages.HEADER_BLOB:
+ log.debug("Decoding 'get blob' response");
+ decodeGetBlobResponse(length, in, out);
+ break;
+ default:
+ log.debug("Invalid type, dropping message");
+ }
+ }
+
+ private void decodeGetHeadResponse(int length, ByteBuf in, List<Object>
out) {
+ byte[] data = new byte[length - 1];
+ in.readBytes(data);
+ String recordId = new String(data, Charsets.UTF_8);
+ out.add(new GetHeadResponse(null, recordId));
+ }
+
+ private void decodeGetSegmentResponse(int length, ByteBuf in, List<Object>
out) {
+ long msb = in.readLong();
+ long lsb = in.readLong();
+
+ String segmentId = new UUID(msb, lsb).toString();
+
+ long hash = in.readLong();
+
+ byte[] data = new byte[length - 25];
+ in.readBytes(data);
+
+ if (hash(data) != hash) {
+ log.debug("Invalid checksum, discarding segment {}", segmentId);
+ return;
+ }
+
+ out.add(new GetSegmentResponse(null, segmentId, data));
+ }
+
+ private void decodeGetBlobResponse(int length, ByteBuf in, List<Object>
out) {
+ int blobIdLength = in.readInt();
+
+ byte[] blobIdBytes = new byte[blobIdLength];
+ in.readBytes(blobIdBytes);
+
+ String blobId = new String(blobIdBytes, Charsets.UTF_8);
+
+ long hash = in.readLong();
+
+ byte[] blobData = new byte[length - 1 - 4 - blobIdBytes.length - 8];
+ in.readBytes(blobData);
+
+ if (hash(blobData) != hash) {
+ log.debug("Invalid checksum, discarding blob {}", blobId);
+ return;
+ }
+
+ out.add(new GetBlobResponse(null, blobId, blobData));
+ }
+
+ private long hash(byte[] data) {
+ return
Hashing.murmur3_32().newHasher().putBytes(data).hash().padToLong();
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/codec/ResponseDecoder.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyBlobReader.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyBlobReader.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyBlobReader.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyBlobReader.java
Thu Sep 15 14:03:05 2016
@@ -17,13 +17,20 @@
package org.apache.jackrabbit.oak.segment.standby.server;
-import org.apache.jackrabbit.oak.api.Blob;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.commons.io.IOUtils;
import org.apache.jackrabbit.oak.plugins.blob.BlobStoreBlob;
import org.apache.jackrabbit.oak.segment.file.FileStore;
import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
class DefaultStandbyBlobReader implements StandbyBlobReader {
+ private final Logger log =
LoggerFactory.getLogger(DefaultStandbyBlobReader.class);
+
private final FileStore store;
DefaultStandbyBlobReader(FileStore store) {
@@ -31,14 +38,22 @@ class DefaultStandbyBlobReader implement
}
@Override
- public Blob readBlob(String blobId) {
+ public byte[] readBlob(String blobId) {
BlobStore blobStore = store.getBlobStore();
- if (blobStore != null) {
- return new BlobStoreBlob(blobStore, blobId);
+ if (blobStore == null) {
+ return null;
+ }
+
+ byte[] bytes = null;
+
+ try (InputStream s = new BlobStoreBlob(blobStore,
blobId).getNewStream()) {
+ bytes = IOUtils.toByteArray(s);
+ } catch (IOException e) {
+ log.warn("Error while reading blob content", e);
}
- return null;
+ return bytes;
}
}
Modified:
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyHeadReader.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyHeadReader.java?rev=1760934&r1=1760933&r2=1760934&view=diff
==============================================================================
---
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyHeadReader.java
(original)
+++
jackrabbit/oak/trunk/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/standby/server/DefaultStandbyHeadReader.java
Thu Sep 15 14:03:05 2016
@@ -17,7 +17,6 @@
package org.apache.jackrabbit.oak.segment.standby.server;
-import org.apache.jackrabbit.oak.segment.RecordId;
import org.apache.jackrabbit.oak.segment.file.FileStore;
class DefaultStandbyHeadReader implements StandbyHeadReader {
@@ -29,8 +28,8 @@ class DefaultStandbyHeadReader implement
}
@Override
- public RecordId readHeadRecordId() {
- return store.getHead().getRecordId();
+ public String readHeadRecordId() {
+ return store.getHead().getRecordId().toString();
}
}
