directory-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From elecha...@apache.org
Subject svn commit: r463366 [3/5] - in /directory/trunks/apacheds/mitosis/src: ./ main/ main/java/ main/java/org/ main/java/org/apache/ main/java/org/apache/directory/ main/java/org/apache/directory/mitosis/ main/java/org/apache/directory/mitosis/common/ main/...
Date Thu, 12 Oct 2006 18:33:19 GMT
Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,84 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.mina.filter.codec.ProtocolDecoderException;
+import org.apache.directory.mitosis.common.CSNVector;
+import org.apache.directory.mitosis.common.ReplicaId;
+import org.apache.directory.mitosis.common.SimpleCSN;
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesAckMessage;
+
+public class BeginLogEntriesAckMessageDecoder extends ResponseMessageDecoder
+{
+    private final CharsetDecoder utf8decoder;
+
+    public BeginLogEntriesAckMessageDecoder()
+    {
+        super( Constants.GET_UPDATE_VECTOR_ACK, 0, 3072 );
+        utf8decoder = Charset.forName( "UTF-8" ).newDecoder();
+    }
+    protected BaseMessage decodeBody( int sequence, int bodyLength,
+                                      int responseCode, ByteBuffer in ) throws Exception
+    {
+        if( responseCode != Constants.OK )
+        {
+            return new BeginLogEntriesAckMessage( sequence, responseCode, null, null );
+        }
+
+        CSNVector purgeVector = new CSNVector();
+        CSNVector updateVector = new CSNVector();
+        BeginLogEntriesAckMessage m = new BeginLogEntriesAckMessage( sequence, responseCode, purgeVector, updateVector );
+        readCSNVector( in, purgeVector );
+        readCSNVector( in, updateVector );
+        
+        return m;
+    }
+
+    private void readCSNVector( ByteBuffer in, CSNVector updateVector ) throws Exception
+    {
+        int nReplicas = in.getInt();
+        if( nReplicas < 0 )
+        {
+            throw new ProtocolDecoderException( "Wrong nReplicas: " + nReplicas );
+        }
+        
+        for( ; nReplicas > 0; nReplicas-- )
+        {
+            ReplicaId replicaId;
+            try
+            {
+                replicaId = new ReplicaId( in.getString( utf8decoder ) );
+            }
+            catch( CharacterCodingException e )
+            {
+                throw new ProtocolDecoderException( "Invalid replicaId", e );
+            }
+            
+            updateVector.setCSN( new SimpleCSN( in.getLong(), replicaId, in.getInt() ) );
+        }
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesAckMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,91 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetEncoder;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.directory.mitosis.common.CSN;
+import org.apache.directory.mitosis.common.CSNVector;
+import org.apache.directory.mitosis.common.ReplicaId;
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesAckMessage;
+
+public class BeginLogEntriesAckMessageEncoder extends ResponseMessageEncoder
+{
+    private final CharsetEncoder utf8encoder;
+
+    public BeginLogEntriesAckMessageEncoder()
+    {
+        utf8encoder = Charset.forName( "UTF-8" ).newEncoder();
+    }
+    
+    protected void encodeBody(BaseMessage in, ByteBuffer out) throws Exception {
+        // write out response code
+        super.encodeBody( in, out );
+        
+        BeginLogEntriesAckMessage m = ( BeginLogEntriesAckMessage ) in;
+        if( m.getResponseCode() != Constants.OK )
+        {
+            return;
+        }
+        
+        writeCSNVector( out, m.getPurgeVector() );
+        writeCSNVector( out, m.getUpdateVector() );
+    }
+
+    private void writeCSNVector( ByteBuffer out, CSNVector csns )
+    {
+        Set replicaIds = csns.getReplicaIds();
+        
+        int nReplicas = replicaIds.size();
+        out.putInt( nReplicas );
+        Iterator it = replicaIds.iterator();
+        while( it.hasNext() )
+        {
+            ReplicaId replicaId = ( ReplicaId ) it.next();
+            CSN csn = csns.getCSN( replicaId );
+            try {
+                out.putString( replicaId.getId(), utf8encoder );
+                out.put( ( byte ) 0x00 );
+                out.putLong( csn.getTimestamp() );
+                out.putInt( csn.getOperationSequence() );
+            }
+            catch ( CharacterCodingException e )
+            {
+                throw new RuntimeException( e );
+            }
+        }
+    }
+
+    public Set getMessageTypes()
+    {
+        Set set = new HashSet();
+        set.add( BeginLogEntriesAckMessage.class );
+        return set;
+    }
+
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class BeginLogEntriesMessageDecoder extends BaseMessageDecoder {
+
+    public BeginLogEntriesMessageDecoder()
+    {
+        super( Constants.BEGIN_LOG_ENTRIES, 0, 0 );
+    }
+
+    protected BaseMessage decodeBody( int sequence, int bodyLength, ByteBuffer in )
+            throws Exception
+    {
+        return new BeginLogEntriesMessage( sequence );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/BeginLogEntriesMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,41 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class BeginLogEntriesMessageEncoder extends BaseMessageEncoder {
+    public BeginLogEntriesMessageEncoder() {
+    }
+
+    protected void encodeBody( BaseMessage in, ByteBuffer out ) {
+    }
+
+    public Set getMessageTypes() {
+        Set set = new HashSet();
+        set.add( BeginLogEntriesMessage.class );
+        return set;
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesAckMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class EndLogEntriesAckMessageDecoder extends ResponseMessageDecoder
+{
+    public EndLogEntriesAckMessageDecoder()
+    {
+        super( Constants.END_LOG_ENTRIES_ACK, 0, 0 );
+    }
+
+    protected BaseMessage decodeBody( int sequence, int bodyLength,
+            int responseCode, ByteBuffer in ) throws Exception
+    {
+        return new EndLogEntriesAckMessage( sequence, responseCode );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesAckMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,38 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesAckMessage;
+
+public class EndLogEntriesAckMessageEncoder extends ResponseMessageEncoder {
+
+    public EndLogEntriesAckMessageEncoder() {
+        super();
+    }
+
+    public Set getMessageTypes() {
+        Set set = new HashSet();
+        set.add( EndLogEntriesAckMessage.class );
+        return set;
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class EndLogEntriesMessageDecoder extends BaseMessageDecoder {
+
+    public EndLogEntriesMessageDecoder()
+    {
+        super( Constants.END_LOG_ENTRIES, 0, 0 );
+    }
+
+    protected BaseMessage decodeBody( int sequence, int bodyLength, ByteBuffer in )
+            throws Exception
+    {
+        return new EndLogEntriesMessage( sequence );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/EndLogEntriesMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,41 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class EndLogEntriesMessageEncoder extends BaseMessageEncoder {
+    public EndLogEntriesMessageEncoder() {
+    }
+
+    protected void encodeBody( BaseMessage in, ByteBuffer out ) {
+    }
+
+    public Set getMessageTypes() {
+        Set set = new HashSet();
+        set.add( EndLogEntriesMessage.class );
+        return set;
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.LogEntryAckMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class LogEntryAckMessageDecoder extends ResponseMessageDecoder
+{
+    public LogEntryAckMessageDecoder()
+    {
+        super( Constants.LOG_ENTRY_ACK, 0, 0 );
+    }
+
+    protected BaseMessage decodeBody( int sequence, int bodyLength,
+            int responseCode, ByteBuffer in ) throws Exception
+    {
+        return new LogEntryAckMessage( sequence, responseCode );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryAckMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.directory.mitosis.service.protocol.message.LogEntryAckMessage;
+
+public class LogEntryAckMessageEncoder extends ResponseMessageEncoder {
+
+    public LogEntryAckMessageEncoder() {
+        super();
+    }
+
+    public Set getMessageTypes() {
+        Set set = new HashSet();
+        set.add( LogEntryAckMessage.class );
+        return set;
+    }
+
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,48 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import org.apache.directory.mitosis.operation.OperationCodec;
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.LogEntryMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class LogEntryMessageDecoder extends BaseMessageDecoder
+{
+    private final OperationCodec operationCodec = new OperationCodec();
+    
+    public LogEntryMessageDecoder()
+    {
+        super( Constants.LOG_ENTRY, 1, Integer.MAX_VALUE );
+    }
+
+    protected BaseMessage decodeBody( int sequence, int bodyLength,
+                                      ByteBuffer in ) throws Exception
+    {
+        byte[] src = new byte[ in.remaining() ];
+        in.get( src );
+        
+        return new LogEntryMessage(
+                sequence,
+                operationCodec.decode( src ) );
+    }
+
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LogEntryMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,51 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.directory.mitosis.operation.OperationCodec;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.LogEntryMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class LogEntryMessageEncoder extends BaseMessageEncoder
+{
+    private final OperationCodec operationCodec = new OperationCodec();
+    
+    public LogEntryMessageEncoder()
+    {
+    }
+
+    protected void encodeBody( BaseMessage in, ByteBuffer out )
+    {
+        LogEntryMessage m = ( LogEntryMessage ) in;
+        out.put( operationCodec.encode( m.getOperation() ) );
+    }
+
+    public Set getMessageTypes()
+    {
+        Set set = new HashSet();
+        set.add( LogEntryMessage.class );
+        return set;
+    }
+
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,46 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.directory.mitosis.common.ReplicaId;
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginAckMessage;
+
+public class LoginAckMessageDecoder extends ResponseMessageDecoder
+{
+    private final CharsetDecoder utf8decoder = Charset.forName("UTF-8").newDecoder();
+
+    public LoginAckMessageDecoder()
+    {
+        super( Constants.LOGIN_ACK, 1, 64 );
+    }
+
+    protected BaseMessage decodeBody( int sequence, int bodyLength,
+                                      int responseCode, ByteBuffer in ) throws Exception
+    {
+        return new LoginAckMessage( sequence, responseCode, new ReplicaId( in.getString( utf8decoder ) ) );
+    }
+
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginAckMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,53 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetEncoder;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginAckMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public class LoginAckMessageEncoder extends ResponseMessageEncoder
+{
+    private final CharsetEncoder utf8encoder = Charset.forName("UTF-8").newEncoder();
+
+    public LoginAckMessageEncoder()
+    {
+    }
+
+    public Set getMessageTypes()
+    {
+        Set set = new HashSet();
+        set.add( LoginAckMessage.class );
+        return set;
+    }
+
+    protected void encodeBody( BaseMessage in, ByteBuffer out ) throws Exception
+    {
+        LoginAckMessage m = ( LoginAckMessage ) in;
+        super.encodeBody( in, out );
+        out.putString( m.getReplicaId().getId(), utf8encoder );
+    }
+
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,46 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.directory.mitosis.common.ReplicaId;
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginMessage;
+
+public class LoginMessageDecoder extends BaseMessageDecoder
+{
+    private final CharsetDecoder utf8decoder; 
+
+    public LoginMessageDecoder()
+    {
+        super( Constants.LOGIN, 0, 32 );
+        utf8decoder = Charset.forName( "UTF-8" ).newDecoder();
+    }
+
+    protected BaseMessage decodeBody( int sequence, int bodyLength,
+                                      ByteBuffer in ) throws Exception
+    {
+        return new LoginMessage( sequence, new ReplicaId( in.getString( utf8decoder ) ) );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/LoginMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,63 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetEncoder;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginMessage;
+import org.apache.mina.common.ByteBuffer;
+
+
+public class LoginMessageEncoder extends BaseMessageEncoder
+{
+    private final CharsetEncoder utf8encoder;
+
+    public LoginMessageEncoder()
+    {
+        utf8encoder = Charset.forName( "UTF-8" ).newEncoder();
+    }
+
+    protected void encodeBody( BaseMessage in, ByteBuffer out )
+    {
+        LoginMessage m = ( LoginMessage ) in;
+        
+        try
+        {
+            out.putString( m.getReplicaId().getId(), utf8encoder );
+        }
+        catch( CharacterCodingException e )
+        {
+            throw new RuntimeException( e );
+        }
+    }
+
+    public Set getMessageTypes() 
+    {
+        Set set = new HashSet();
+        set.add( LoginMessage.class );
+        
+        return set;
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationClientProtocolCodecFactory.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationClientProtocolCodecFactory.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationClientProtocolCodecFactory.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationClientProtocolCodecFactory.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import org.apache.mina.filter.codec.demux.DemuxingProtocolCodecFactory;
+
+public class ReplicationClientProtocolCodecFactory extends
+        DemuxingProtocolCodecFactory {
+
+    public ReplicationClientProtocolCodecFactory()
+    {
+        register( LogEntryMessageEncoder.class );
+        register( LoginMessageEncoder.class );
+        register( BeginLogEntriesMessageEncoder.class );
+        register( EndLogEntriesMessageEncoder.class );
+        
+        register( LogEntryAckMessageDecoder.class );
+        register( LoginAckMessageDecoder.class );
+        register( BeginLogEntriesAckMessageDecoder.class );
+        register( EndLogEntriesAckMessageDecoder.class );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationServerProtocolCodecFactory.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationServerProtocolCodecFactory.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationServerProtocolCodecFactory.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ReplicationServerProtocolCodecFactory.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import org.apache.mina.filter.codec.demux.DemuxingProtocolCodecFactory;
+
+public class ReplicationServerProtocolCodecFactory extends
+        DemuxingProtocolCodecFactory {
+
+    public ReplicationServerProtocolCodecFactory()
+    {
+        register( LogEntryAckMessageEncoder.class );
+        register( LoginAckMessageEncoder.class );
+        register( BeginLogEntriesAckMessageEncoder.class );
+        register( EndLogEntriesAckMessageEncoder.class );
+        
+        register( LogEntryMessageDecoder.class );
+        register( LoginMessageDecoder.class );
+        register( BeginLogEntriesMessageDecoder.class );
+        register( EndLogEntriesMessageDecoder.class );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageDecoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageDecoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageDecoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageDecoder.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,42 @@
+/*
+ *  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.directory.mitosis.service.protocol.codec;
+
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public abstract class ResponseMessageDecoder extends BaseMessageDecoder
+{
+
+    protected ResponseMessageDecoder( int type, int minBodyLength, int maxBodyLength )
+    {
+        super( type, minBodyLength + 4, maxBodyLength + 4 );
+    }
+
+    protected final BaseMessage decodeBody( int sequence, int bodyLength,
+                                      ByteBuffer in ) throws Exception
+    {
+        return decodeBody( sequence, bodyLength, in.getInt(), in );
+    }
+
+    protected abstract BaseMessage decodeBody( int sequence, int bodyLength,
+                                               int responseCode, ByteBuffer in ) throws Exception;
+
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageEncoder.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageEncoder.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageEncoder.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/codec/ResponseMessageEncoder.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.codec;
+
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.ResponseMessage;
+import org.apache.mina.common.ByteBuffer;
+
+public abstract class ResponseMessageEncoder extends BaseMessageEncoder
+{
+    public ResponseMessageEncoder()
+    {
+    }
+
+    protected void encodeBody( BaseMessage in, ByteBuffer out ) throws Exception
+    {
+        ResponseMessage m = ( ResponseMessage ) in;
+        out.putInt( m.getResponseCode() );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientContextHandler.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientContextHandler.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientContextHandler.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientContextHandler.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,398 @@
+/*
+ *  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.directory.mitosis.service.protocol.handler;
+
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+
+import javax.naming.Name;
+import javax.naming.NamingEnumeration;
+import javax.naming.NamingException;
+import javax.naming.directory.Attribute;
+import javax.naming.directory.Attributes;
+import javax.naming.directory.SearchControls;
+import javax.naming.directory.SearchResult;
+
+import org.apache.directory.shared.ldap.filter.PresenceNode;
+import org.apache.directory.shared.ldap.name.LdapDN;
+import org.apache.mina.common.IdleStatus;
+import org.apache.mina.util.SessionLog;
+import org.apache.directory.mitosis.common.CSN;
+import org.apache.directory.mitosis.common.CSNVector;
+import org.apache.directory.mitosis.common.Replica;
+import org.apache.directory.mitosis.common.ReplicaId;
+import org.apache.directory.mitosis.common.SimpleCSN;
+import org.apache.directory.mitosis.operation.AddEntryOperation;
+import org.apache.directory.mitosis.operation.Operation;
+import org.apache.directory.mitosis.service.ReplicationContext;
+import org.apache.directory.mitosis.service.ReplicationContext.State;
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BaseMessage;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesMessage;
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesMessage;
+import org.apache.directory.mitosis.service.protocol.message.LogEntryAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.LogEntryMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginMessage;
+import org.apache.directory.mitosis.store.ReplicationLogIterator;
+import org.apache.directory.mitosis.store.ReplicationStore;
+
+/**
+ * {@link ReplicationContextHandler} that implements client-side replication logic
+ * which sends any changes out-of-date to server.
+ *
+ * @author Trustin Lee
+ * @version $Rev: 116 $, $Date: 2006-09-18 13:47:53Z $
+ */
+public class ReplicationClientContextHandler implements
+        ReplicationContextHandler
+{
+    public void contextBegin( ReplicationContext ctx ) throws Exception
+    {
+        // Send a login message.
+        LoginMessage m = new LoginMessage(
+                ctx.getNextSequence(),
+                ctx.getService().getConfiguration().getReplicaId() );
+        ctx.getSession().write( m );
+        
+        // Set write timeout
+        ctx.getSession().setWriteTimeout( ctx.getConfiguration().getResponseTimeout() );
+        
+        // Check update vector of the remote peer every 5 seconds.
+        ctx.getSession().setIdleTime( IdleStatus.BOTH_IDLE, 5 );
+    }
+
+    public void contextEnd( ReplicationContext ctx ) throws Exception
+    {
+    }
+
+    public void messageReceived( ReplicationContext ctx, Object message )
+            throws Exception
+    {
+        ctx.cancelExpiration( ( ( BaseMessage ) message ).getSequence() );
+        
+        if( ctx.getState() == State.READY )
+        {
+            if( message instanceof LogEntryAckMessage )
+            {
+                onLogEntryAck( ctx, ( LogEntryAckMessage ) message );
+            }
+            else if( message instanceof BeginLogEntriesAckMessage )
+            {
+                onBeginLogEntriesAck( ctx, ( BeginLogEntriesAckMessage ) message );
+            }
+            else if( message instanceof EndLogEntriesAckMessage )
+            {
+                // Do nothing
+            }
+            else
+            {
+                onUnexpectedMessage( ctx, message );
+            }
+        }
+        else
+        {
+            if( message instanceof LoginAckMessage )
+            {
+                onLoginAck( ctx, ( LoginAckMessage ) message );
+            }
+            else
+            {
+                onUnexpectedMessage( ctx, message );
+            }
+        }
+    }
+
+    public void messageSent( ReplicationContext ctx, Object message )
+            throws Exception
+    {
+        if( message instanceof LogEntryMessage ||
+                message instanceof LoginMessage )
+        {
+            ctx.scheduleExpiration( message );
+        }
+    }
+
+    public void exceptionCaught( ReplicationContext ctx, Throwable cause )
+            throws Exception
+    {
+        SessionLog.warn( ctx.getSession(), "Unexpected exception.", cause );
+        ctx.getSession().close();
+    }
+
+    public void contextIdle( ReplicationContext ctx, IdleStatus status )
+            throws Exception
+    {
+        // If this cilent is logged in, all responses for sent messages
+        // (LogEntryMessages) is received, and no write request is pending,
+        // it means previous replication process ended or this is the
+        // first replication attempt.
+        if( ctx.getState() == State.READY &&
+                ctx.getScheduledExpirations() == 0 &&
+                ctx.getSession().getScheduledWriteRequests() == 0 )
+        {
+            beginReplication( ctx );
+        }
+    }
+
+    private void onLoginAck( ReplicationContext ctx, LoginAckMessage message )
+    {
+        if( message.getResponseCode() != Constants.OK )
+        {
+            SessionLog.warn(
+                    ctx.getSession(),
+                    "Login attempt failed: " + message.getResponseCode() );
+            ctx.getSession().close();
+            return;
+        }
+        
+        Iterator i = ctx.getConfiguration().getPeerReplicas().iterator();
+        while( i.hasNext() )
+        {
+            Replica replica = ( Replica ) i.next();
+            if( replica.getId().equals( message.getReplicaId() ) )
+            {
+                if( replica.getAddress().getAddress().equals(
+                        ( ( InetSocketAddress ) ctx.getSession().getRemoteAddress() ).getAddress() ) )
+                {
+                    ctx.setPeer( replica );
+                    ctx.setState( State.READY );
+                    
+                    beginReplication( ctx );
+                    return;
+                }
+                else
+                {
+                    SessionLog.warn(
+                            ctx.getSession(),
+                            "Peer address mismatches: " + 
+                            ctx.getSession().getRemoteAddress() + 
+                            " (expected: " + replica.getAddress() );
+                    ctx.getSession().close();
+                    return;
+                }
+            }
+        }
+        
+        SessionLog.warn(
+                ctx.getSession(),
+                "Unknown peer replica ID: " + message.getReplicaId() );
+        ctx.getSession().close();
+    }
+
+    private void beginReplication( ReplicationContext ctx )
+    {
+        // Initiate replication process asking update vector.
+        ctx.getSession().write( new BeginLogEntriesMessage( ctx.getNextSequence() ) );
+    }
+
+    private void onLogEntryAck( ReplicationContext ctx, LogEntryAckMessage message ) throws Exception
+    {
+        if( message.getResponseCode() != Constants.OK )
+        {
+            SessionLog.warn(
+                    ctx.getSession(),
+                    "Remote peer failed to execute a log entry." );
+            ctx.getSession().close();
+        }
+    }
+
+    private void onBeginLogEntriesAck( ReplicationContext ctx, BeginLogEntriesAckMessage message ) throws NamingException
+    {
+        // Start transaction only when the server says OK.
+        if( message.getResponseCode() != Constants.OK )
+        {
+            return;
+        }
+
+        ReplicationStore store = ctx.getConfiguration().getStore();
+        CSNVector yourUV = message.getUpdateVector();
+        CSNVector myPV;
+        try
+        {
+            myPV = store.getPurgeVector();
+        }
+        catch( Exception e )
+        {
+            SessionLog.warn(
+                    ctx.getSession(),
+                    "Failed to get update vector.", e );
+            ctx.getSession().close();
+            return;
+        }
+        
+        // Do full-DIT transfer if the peer is new and I'm not new.
+        try
+        {
+            if( myPV.size() > 0 && yourUV.size() == 0 )
+            {
+                SessionLog.warn(
+                        ctx.getSession(),
+                        "Starting a whole DIT transfer." );
+                sendAllEntries( ctx );
+            }
+            else
+            {
+                SessionLog.warn(
+                        ctx.getSession(),
+                        "Starting a partial replication log transfer." );
+                sendReplicationLogs( ctx, myPV, yourUV );
+            }
+        }
+        finally
+        {
+            // Send EngLogEntries message to release the remote peer resources.
+            ctx.getSession().write( new EndLogEntriesMessage ( ctx.getNextSequence() ) );
+        }
+    }
+    
+    private void sendAllEntries( ReplicationContext ctx ) throws NamingException
+    {
+        Attributes rootDSE =
+            ctx.getServiceConfiguration().getPartitionNexus().getRootDSE();
+        
+        Attribute namingContextsAttr = rootDSE.get( "namingContexts" );
+        if( namingContextsAttr == null || namingContextsAttr.size() == 0 )
+        {
+            SessionLog.warn(
+                    ctx.getSession(),
+                    "No namingContexts attributes in rootDSE." );
+            return;
+        }
+        
+        // Iterate all context partitions to send all entries of them.
+        NamingEnumeration e = namingContextsAttr.getAll();
+        while( e.hasMore() )
+        {
+            Object value = e.next();
+            // Convert attribute value to JNDI name.
+            Name contextName;
+            if( value instanceof Name )
+            {
+                contextName = ( Name ) value;
+            }
+            else
+            {
+                contextName = new LdapDN( String.valueOf( value ) );
+            }
+            
+            SessionLog.info(
+                    ctx.getSession(),
+                    "Sending entries under '" + contextName + '\'' );
+            sendAllEntries( ctx, contextName );
+        }
+    }
+
+    private void sendAllEntries( ReplicationContext ctx, Name contextName ) throws NamingException
+    {
+        // Retrieve all subtree including the base entry
+        SearchControls ctrl = new SearchControls();
+        ctrl.setSearchScope( SearchControls.SUBTREE_SCOPE ); 
+        NamingEnumeration e = ctx.getServiceConfiguration().getPartitionNexus().search(
+                (LdapDN)contextName,
+                ctx.getServiceConfiguration().getEnvironment(),
+                new PresenceNode( "objectClass" ), ctrl );
+        
+        try
+        {
+            while( e.hasMore() )
+            {
+                SearchResult sr = ( SearchResult ) e.next();
+                Attributes attrs = sr.getAttributes();
+                
+                // Skip entries without entryCSN attribute.
+                Attribute entryCSNAttr = attrs.get( org.apache.directory.mitosis.common.Constants.ENTRY_CSN );
+                if( entryCSNAttr == null )
+                {
+                    continue;
+                }
+                
+                // Get entryCSN of the entry.  Skip if entryCSN value is invalid. 
+                CSN csn = null;
+                try
+                {
+                    csn = new SimpleCSN( String.valueOf( entryCSNAttr.get() ) );
+                }
+                catch( IllegalArgumentException ex )
+                {
+                    SessionLog.warn(
+                            ctx.getSession(),
+                            "An entry with improper entryCSN: " + sr.getName() );
+                    continue;
+                }
+                
+                // Convert the entry into AddEntryOperation log.
+                Operation op = new AddEntryOperation(
+                        csn,
+                        new LdapDN( sr.getName() ), sr.getName(), attrs );
+                
+                // Send a LogEntry message for the entry.
+                ctx.getSession().write( new LogEntryMessage( ctx.getNextSequence(), op ) );
+            }
+        }
+        finally
+        {
+            e.close();
+        }
+    }
+
+    private void sendReplicationLogs( ReplicationContext ctx, CSNVector myPV, CSNVector yourUV )
+    {
+        Iterator i = myPV.getReplicaIds().iterator();
+        while( i.hasNext() )
+        {
+            ReplicaId replicaId = ( ReplicaId ) i.next();
+            CSN myCSN = myPV.getCSN( replicaId );
+            CSN yourCSN = yourUV.getCSN( replicaId );
+            if( yourCSN != null && ( myCSN == null || yourCSN.compareTo( myCSN ) < 0 ) )
+            {
+                SessionLog.warn(
+                        ctx.getSession(),
+                        "Remote update vector (" + yourUV + ") is out-of-date.  Full replication is required." );
+                ctx.getSession().close();
+                return;
+            }
+        }
+        
+        ReplicationLogIterator logIt = ctx.getConfiguration().getStore().getLogs( yourUV, false );
+        try
+        {
+            while( logIt.next() )
+            {
+                Operation op = logIt.getOperation();
+                ctx.getSession().write( new LogEntryMessage( ctx.getNextSequence(), op ) );
+            }
+        }
+        finally
+        {
+            logIt.close();
+        }
+    }
+    
+    private void onUnexpectedMessage( ReplicationContext ctx, Object message )
+    {
+        SessionLog.warn(
+                ctx.getSession(),
+                "Unexpected message: " + message );
+        ctx.getSession().close();
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientProtocolHandler.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientProtocolHandler.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientProtocolHandler.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationClientProtocolHandler.java Thu Oct 12 11:33:14 2006
@@ -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.directory.mitosis.service.protocol.handler;
+
+import org.apache.directory.mitosis.service.ReplicationService;
+import org.apache.directory.mitosis.service.protocol.codec.ReplicationClientProtocolCodecFactory;
+
+public class ReplicationClientProtocolHandler extends
+        ReplicationProtocolHandler
+{
+    public ReplicationClientProtocolHandler( ReplicationService service )
+    {
+        super( service, new ReplicationClientContextHandler(), new ReplicationClientProtocolCodecFactory() );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationContextHandler.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationContextHandler.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationContextHandler.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationContextHandler.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,33 @@
+/*
+ *  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.directory.mitosis.service.protocol.handler;
+
+import org.apache.directory.mitosis.service.ReplicationContext;
+import org.apache.mina.common.IdleStatus;
+
+
+public interface ReplicationContextHandler {
+    void contextBegin( ReplicationContext ctx ) throws Exception;
+    void contextEnd( ReplicationContext ctx ) throws Exception;
+    void messageReceived( ReplicationContext ctx, Object message ) throws Exception;
+    void messageSent( ReplicationContext ctx, Object message ) throws Exception;
+    void exceptionCaught( ReplicationContext ctx, Throwable cause ) throws Exception;
+    void contextIdle( ReplicationContext ctx, IdleStatus status ) throws Exception;
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationProtocolHandler.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationProtocolHandler.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationProtocolHandler.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationProtocolHandler.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,105 @@
+/*
+ *  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.directory.mitosis.service.protocol.handler;
+
+import org.apache.directory.mitosis.configuration.ReplicationConfiguration;
+import org.apache.directory.mitosis.service.ReplicationContext;
+import org.apache.directory.mitosis.service.ReplicationService;
+import org.apache.directory.mitosis.service.SimpleReplicationContext;
+import org.apache.directory.server.core.DirectoryServiceConfiguration;
+import org.apache.mina.common.IdleStatus;
+import org.apache.mina.common.IoHandler;
+import org.apache.mina.common.IoSession;
+import org.apache.mina.filter.LoggingFilter;
+import org.apache.mina.filter.codec.ProtocolCodecFactory;
+import org.apache.mina.filter.codec.ProtocolCodecFilter;
+
+public class ReplicationProtocolHandler implements IoHandler
+{
+    private static final String CONTEXT = "context";
+    
+    private final ReplicationService service;
+    private final ReplicationConfiguration configuration;
+    private final DirectoryServiceConfiguration serviceCfg;
+    private final ReplicationContextHandler contextHandler;
+    private final ProtocolCodecFactory codecFactory;
+
+    public ReplicationProtocolHandler(
+            ReplicationService service,
+            ReplicationContextHandler contextHandler,
+            ProtocolCodecFactory codecFactory )
+    {
+        assert service != null;
+        assert contextHandler != null;
+        assert codecFactory != null;
+
+        this.service = service;
+        this.configuration = service.getConfiguration();
+        this.serviceCfg = service.getFactoryConfiguration();
+        this.contextHandler = contextHandler;
+        this.codecFactory = codecFactory;
+    }
+    
+    private ReplicationContext getContext( IoSession session )
+    {
+        return ( ReplicationContext ) session.getAttribute( CONTEXT );
+    }
+    
+    public void sessionCreated( IoSession session ) throws Exception
+    {
+        session.setAttribute( CONTEXT, new SimpleReplicationContext( service, serviceCfg, configuration, session ) );
+        session.getFilterChain().addLast(
+                "codec", new ProtocolCodecFilter( codecFactory ) );
+        session.getFilterChain().addLast(
+                "log", new LoggingFilter() );
+    }
+
+    public void exceptionCaught( IoSession session, Throwable cause ) throws Exception
+    {
+        contextHandler.exceptionCaught( getContext( session ), cause );
+    }
+
+    public void messageReceived( IoSession session, Object message ) throws Exception
+    {
+        contextHandler.messageReceived( getContext( session ), message );
+    }
+
+    public void messageSent( IoSession session, Object message ) throws Exception
+    {
+        contextHandler.messageSent( getContext( session ), message );
+    }
+
+    public void sessionClosed( IoSession session ) throws Exception
+    {
+        ReplicationContext ctx = getContext( session );
+        contextHandler.contextEnd( ctx );
+        ctx.cancelAllExpirations();
+    }
+
+    public void sessionIdle( IoSession session, IdleStatus status ) throws Exception
+    {
+        contextHandler.contextIdle( getContext( session ), status );
+    }
+
+    public void sessionOpened( IoSession session ) throws Exception
+    {
+        contextHandler.contextBegin( getContext( session ) );
+    }
+}

Added: directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationServerContextHandler.java
URL: http://svn.apache.org/viewvc/directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationServerContextHandler.java?view=auto&rev=463366
==============================================================================
--- directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationServerContextHandler.java (added)
+++ directory/trunks/apacheds/mitosis/src/main/java/org/apache/directory/mitosis/service/protocol/handler/ReplicationServerContextHandler.java Thu Oct 12 11:33:14 2006
@@ -0,0 +1,266 @@
+/*
+ *  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.directory.mitosis.service.protocol.handler;
+
+import java.net.InetSocketAddress;
+import java.util.Iterator;
+
+import org.apache.mina.common.IdleStatus;
+import org.apache.mina.util.SessionLog;
+import org.apache.directory.mitosis.common.CSNVector;
+import org.apache.directory.mitosis.common.Replica;
+import org.apache.directory.mitosis.operation.Operation;
+import org.apache.directory.mitosis.service.ReplicationContext;
+import org.apache.directory.mitosis.service.ReplicationContext.State;
+import org.apache.directory.mitosis.service.protocol.Constants;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.BeginLogEntriesMessage;
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.EndLogEntriesMessage;
+import org.apache.directory.mitosis.service.protocol.message.LogEntryAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.LogEntryMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginAckMessage;
+import org.apache.directory.mitosis.service.protocol.message.LoginMessage;
+import org.apache.directory.mitosis.store.ReplicationStore;
+
+/**
+ * {@link ReplicationContextHandler} that implements server-side replication logic
+ * which retrieves any changes occurred in remote replicas.
+ *
+ * @author Trustin Lee
+ * @version $Rev: 116 $, $Date: 2006-09-18 13:47:53Z $
+ */
+public class ReplicationServerContextHandler implements
+        ReplicationContextHandler
+{
+    private Replica replicaInTransaction = null;
+    
+    public void contextBegin( ReplicationContext ctx ) throws Exception
+    {
+        // Set login timeout
+        ctx.getSession().setIdleTime(
+                IdleStatus.BOTH_IDLE,
+                ctx.getConfiguration().getResponseTimeout() );
+        
+        // Set write timeout
+        ctx.getSession().setWriteTimeout( ctx.getConfiguration().getResponseTimeout() );
+    }
+
+    public synchronized void contextEnd( ReplicationContext ctx ) throws Exception
+    {
+        // Reset the mark if the context has the unfinished transaction.
+        if( !ctx.getPeer().equals( replicaInTransaction ) )
+        {
+            replicaInTransaction = null;
+        }
+    }
+
+    public void messageReceived( ReplicationContext ctx, Object message )
+            throws Exception
+    {
+        if( ctx.getState() == State.READY )
+        {
+            if( message instanceof LogEntryMessage )
+            {
+                onLogEntry( ctx, ( LogEntryMessage ) message );
+            }
+            else if( message instanceof BeginLogEntriesMessage )
+            {
+                onBeginLogEntries( ctx, ( BeginLogEntriesMessage ) message );
+            }
+            else if( message instanceof EndLogEntriesMessage )
+            {
+                onEndLogEntries( ctx, ( EndLogEntriesMessage ) message );
+            }
+            else
+            {
+                onUnexpectedMessage( ctx, message );
+            }
+        }
+        else
+        {
+            if( message instanceof LoginMessage )
+            {
+                onLogin( ctx, ( LoginMessage ) message );
+            }
+            else
+            {
+                onUnexpectedMessage( ctx, message );
+            }
+        }
+    }
+
+    public void messageSent( ReplicationContext ctx, Object message )
+            throws Exception
+    {
+    }
+
+    public void exceptionCaught( ReplicationContext ctx, Throwable cause )
+            throws Exception
+    {
+        SessionLog.warn( ctx.getSession(), "Unexpected exception.", cause );
+        ctx.getSession().close();
+    }
+
+    public void contextIdle( ReplicationContext ctx, IdleStatus status )
+            throws Exception
+    {
+        if( ctx.getState() == State.INIT )
+        {
+            SessionLog.warn(
+                    ctx.getSession(),
+                    "No login attempt in " +
+                    ctx.getConfiguration().getResponseTimeout() +
+                    " second(s)." );
+            ctx.getSession().close();
+        }
+    }
+
+    private void onLogin( ReplicationContext ctx, LoginMessage message )
+    {
+        Iterator i = ctx.getConfiguration().getPeerReplicas().iterator();
+        while( i.hasNext() )
+        {
+            Replica replica = ( Replica ) i.next();
+            if( replica.getId().equals( message.getReplicaId() ) )
+            {
+                if( replica.getAddress().getAddress().equals(
+                        ( ( InetSocketAddress ) ctx.getSession().getRemoteAddress() ).getAddress() ) )
+                {
+                    ctx.getSession().write(
+                            new LoginAckMessage(
+                                    message.getSequence(),
+                                    Constants.OK,
+                                    ctx.getConfiguration().getReplicaId() ) );
+                    ctx.setPeer( replica );
+                    ctx.setState( State.READY );
+                    
+                    // Clear login timeout.
+                    ctx.getSession().setIdleTime( IdleStatus.BOTH_IDLE, 0 );
+                    return;
+                }
+                else
+                {
+                    SessionLog.warn(
+                            ctx.getSession(),
+                            "Peer address mismatches: " + 
+                            ctx.getSession().getRemoteAddress() + 
+                            " (expected: " + replica.getAddress() );
+                    ctx.getSession().write(
+                            new LoginAckMessage(
+                                    message.getSequence(),
+                                    Constants.NOT_OK,
+                                    ctx.getConfiguration().getReplicaId() ) );
+                    ctx.getSession().close();
+                    return;
+                }
+            }
+        }
+        
+        SessionLog.warn(
+                ctx.getSession(),
+                "Unknown peer replica ID: " + message.getReplicaId() );
+        ctx.getSession().write(
+                new LoginAckMessage(
+                        message.getSequence(),
+                        Constants.NOT_OK,
+                        ctx.getConfiguration().getReplicaId() ) );
+        ctx.getSession().close();
+    }
+
+    private synchronized void onLogEntry( ReplicationContext ctx, LogEntryMessage message ) throws Exception
+    {
+        // Return error if other replica than what is in progress sends
+        // a log entry
+        if( !ctx.getPeer().equals( replicaInTransaction ) )
+        {
+            ctx.getSession().write( new LogEntryAckMessage( message.getSequence(), Constants.NOT_OK ) );
+            return;
+        }
+            
+        Operation op = message.getOperation();
+        LogEntryAckMessage ack = null;
+        try
+        {
+            op.execute( ctx.getServiceConfiguration().getPartitionNexus(), ctx.getConfiguration().getStore() );
+            ack = new LogEntryAckMessage( message.getSequence(), Constants.OK ); 
+        }
+        catch( Exception e )
+        {
+            ack = new LogEntryAckMessage( message.getSequence(), Constants.NOT_OK );
+            throw e;
+        }
+        finally
+        {
+            ctx.getSession().write( ack );
+        }
+    }
+
+    private synchronized void onBeginLogEntries( ReplicationContext ctx, BeginLogEntriesMessage message )
+    {
+        // Return error if the transaction is already in progress.
+        if( replicaInTransaction != null )
+        {
+            ctx.getSession().write( new BeginLogEntriesAckMessage( message.getSequence(), Constants.NOT_OK, null, null ) );
+            return;
+        }
+
+        ReplicationStore store = ctx.getConfiguration().getStore();
+        try
+        {
+            CSNVector pv = store.getPurgeVector();
+            CSNVector uv = store.getUpdateVector();
+            replicaInTransaction = ctx.getPeer(); // Mark as replica in transaction
+            ctx.getSession().write( new BeginLogEntriesAckMessage(
+                    message.getSequence(), Constants.OK, pv, uv ) );
+        }
+        catch( Exception e )
+        {
+            SessionLog.warn(
+                    ctx.getSession(),
+                    "Failed to get update vector.", e );
+            ctx.getSession().write(
+                    new BeginLogEntriesAckMessage(
+                            message.getSequence(), Constants.NOT_OK, null, null ) );
+        }
+    }
+
+    private synchronized void onEndLogEntries( ReplicationContext ctx, EndLogEntriesMessage message )
+    {
+        // Return error if other replica than what is in progress sends
+        // a flow control message
+        if( !ctx.getPeer().equals( replicaInTransaction ) )
+        {
+            ctx.getSession().write( new EndLogEntriesAckMessage( message.getSequence(), Constants.NOT_OK ) );
+            return;
+        }
+        
+        ctx.getSession().write( new EndLogEntriesAckMessage( message.getSequence(), Constants.OK ) );
+        replicaInTransaction = null; // Reset the mark.
+    }
+
+    private void onUnexpectedMessage( ReplicationContext ctx, Object message )
+    {
+        SessionLog.warn(
+                ctx.getSession(),
+                "Unexpected message: " + message );
+        ctx.getSession().close();
+    }
+}



Mime
View raw message