ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [07/50] incubator-ignite git commit: # IGNITE-61 - Fixing client
Date Tue, 10 Feb 2015 12:37:11 GMT
# IGNITE-61 - Fixing client


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/b9cb1573
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/b9cb1573
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/b9cb1573

Branch: refs/heads/ignite-96-tests
Commit: b9cb15731239bb086895fcf1b7190317f90a483d
Parents: d64fefb
Author: Valentin Kulichenko <vkulichenko@gridgain.com>
Authored: Mon Feb 9 00:28:35 2015 -0800
Committer: Valentin Kulichenko <vkulichenko@gridgain.com>
Committed: Mon Feb 9 00:28:35 2015 -0800

----------------------------------------------------------------------
 .../GridClientConnectionManagerAdapter.java     |  84 +--
 .../connection/GridClientNioTcpConnection.java  | 135 +----
 .../communication/GridIoMessageFactory.java     |  21 -
 .../GridClientHandshakeRequestWrapper.java      | 117 -----
 .../GridClientHandshakeResponseWrapper.java     |  94 ----
 .../message/GridClientMessageWrapper.java       | 275 ----------
 .../message/GridClientPingPacketWrapper.java    |  89 ----
 .../protocols/tcp/GridClientPacketType.java     |   3 +
 .../protocols/tcp/GridMemcachedMessage.java     |   3 +
 .../protocols/tcp/GridTcpRestDirectParser.java  | 523 -------------------
 .../protocols/tcp/GridTcpRestNioListener.java   |  29 +-
 .../rest/protocols/tcp/GridTcpRestParser.java   |  26 +
 .../rest/protocols/tcp/GridTcpRestProtocol.java |   8 +-
 13 files changed, 72 insertions(+), 1335 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientConnectionManagerAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientConnectionManagerAdapter.java b/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientConnectionManagerAdapter.java
index 476d9c5..6cecef0 100644
--- a/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientConnectionManagerAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientConnectionManagerAdapter.java
@@ -18,18 +18,17 @@
 package org.apache.ignite.client.impl.connection;
 
 import org.apache.ignite.*;
-import org.apache.ignite.internal.direct.*;
-import org.apache.ignite.logger.java.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-import org.apache.ignite.plugin.security.*;
 import org.apache.ignite.client.*;
 import org.apache.ignite.client.impl.*;
 import org.apache.ignite.client.util.*;
 import org.apache.ignite.internal.processors.rest.client.message.*;
+import org.apache.ignite.internal.processors.rest.protocols.tcp.*;
 import org.apache.ignite.internal.util.nio.*;
 import org.apache.ignite.internal.util.nio.ssl.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.logger.java.*;
+import org.apache.ignite.plugin.security.*;
 import org.jetbrains.annotations.*;
 
 import javax.net.ssl.*;
@@ -143,12 +142,12 @@ abstract class GridClientConnectionManagerAdapter implements GridClientConnectio
 
                 GridNioFilter[] filters;
 
-                GridNioFilter codecFilter = new GridNioCodecFilter(new NioParser(), gridLog, true);
+                GridNioFilter codecFilter = new GridNioCodecFilter(new GridTcpRestParser(), gridLog, false);
 
                 if (sslCtx != null) {
                     GridNioSslFilter sslFilter = new GridNioSslFilter(sslCtx, gridLog);
 
-                    sslFilter.directMode(true);
+                    sslFilter.directMode(false);
                     sslFilter.clientMode(true);
 
                     filters = new GridNioFilter[]{codecFilter, sslFilter};
@@ -166,7 +165,7 @@ abstract class GridClientConnectionManagerAdapter implements GridClientConnectio
                     .byteOrder(ByteOrder.nativeOrder())
                     .tcpNoDelay(cfg.isTcpNoDelay())
                     .directBuffer(true)
-                    .directMode(true)
+                    .directMode(false)
                     .socketReceiveBufferSize(0)
                     .socketSendBufferSize(0)
                     .idleTimeout(Long.MAX_VALUE)
@@ -594,22 +593,10 @@ abstract class GridClientConnectionManagerAdapter implements GridClientConnectio
 
                 assert conn != null;
 
-                if (msg instanceof GridClientMessageWrapper) {
-                    GridClientMessageWrapper req = (GridClientMessageWrapper)msg;
-
-                    if (req.messageSize() != 0) {
-                        assert req.message() != null;
-
-                        conn.handleResponse(req);
-                    }
-                    else
-                        conn.handlePingResponse();
-                }
-                else {
-                    assert msg instanceof GridClientPingPacket : msg;
-
+                if (msg instanceof GridClientPingPacket)
                     conn.handlePingResponse();
-                }
+                else
+                    conn.handleResponse((GridClientMessage)msg);
             }
         }
 
@@ -647,57 +634,4 @@ abstract class GridClientConnectionManagerAdapter implements GridClientConnectio
             ses.close();
         }
     }
-
-    /**
-     *
-     */
-    private static class NioParser implements GridNioParser {
-        /** Message metadata key. */
-        private static final int MSG_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
-
-        /** {@inheritDoc} */
-        @Nullable @Override public Object decode(GridNioSession ses, ByteBuffer buf)
-            throws IOException, IgniteCheckedException {
-            GridClientFutureAdapter<?> handshakeFut = ses.meta(GridClientNioTcpConnection.SES_META_HANDSHAKE);
-
-            if (handshakeFut != null) {
-                byte code = buf.get();
-
-                return new GridClientHandshakeResponse(code);
-            }
-
-            MessageAdapter msg = ses.removeMeta(MSG_META_KEY);
-
-            if (msg == null && buf.hasRemaining()) {
-                byte type = buf.get();
-
-                if (type == GridClientMessageWrapper.REQ_HEADER) {
-                    msg = new GridClientMessageWrapper();
-
-                    msg.setReader(new DirectMessageReader(null));
-                }
-                else
-                    throw new IOException("Invalid message type: " + type);
-            }
-
-            boolean finished = false;
-
-            if (buf.hasRemaining())
-                finished = msg.readFrom(buf);
-
-            if (finished)
-                return msg;
-            else {
-                ses.addMeta(MSG_META_KEY, msg);
-
-                return null;
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override public ByteBuffer encode(GridNioSession ses, Object msg) throws IOException, IgniteCheckedException {
-            // No encoding needed for direct messages.
-            throw new UnsupportedEncodingException();
-        }
-    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientNioTcpConnection.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientNioTcpConnection.java b/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientNioTcpConnection.java
index 51da3d7..3cead45 100644
--- a/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientNioTcpConnection.java
+++ b/modules/core/src/main/java/org/apache/ignite/client/impl/connection/GridClientNioTcpConnection.java
@@ -33,7 +33,6 @@ import org.jetbrains.annotations.*;
 import javax.net.ssl.*;
 import java.io.*;
 import java.net.*;
-import java.nio.*;
 import java.nio.channels.*;
 import java.util.*;
 import java.util.concurrent.*;
@@ -43,6 +42,7 @@ import java.util.logging.*;
 import static org.apache.ignite.client.GridClientCacheFlag.*;
 import static org.apache.ignite.client.impl.connection.GridClientConnectionCloseReason.*;
 import static org.apache.ignite.internal.processors.rest.client.message.GridClientCacheRequest.GridCacheOperation.*;
+import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.*;
 
 /**
  * This class performs request to grid over tcp protocol. Serialization is performed with marshaller
@@ -194,9 +194,9 @@ public class GridClientNioTcpConnection extends GridClientConnection {
             else if (marsh instanceof GridClientJdkMarshaller)
                 req.marshallerId(GridClientJdkMarshaller.ID);
 
-            GridClientHandshakeRequestWrapper wrapper = new GridClientHandshakeRequestWrapper(req);
+            ses.addMeta(MARSHALLER.ordinal(), marsh);
 
-            ses.send(wrapper);
+            ses.send(req);
 
             handshakeFut.get();
 
@@ -392,7 +392,7 @@ public class GridClientNioTcpConnection extends GridClientConnection {
             else if (now - lastPingSndTime > pingInterval && lastPingRcvTime != Long.MAX_VALUE) {
                 lastPingRcvTime = Long.MAX_VALUE;
 
-                ses.send(new GridClientPingPacketWrapper());
+                ses.send(GridClientPingPacket.PING_MESSAGE);
 
                 lastPingSndTime = now;
             }
@@ -415,22 +415,7 @@ public class GridClientNioTcpConnection extends GridClientConnection {
 
             assert old == null;
 
-            GridClientMessageWrapper wrapper;
-
-            try {
-                wrapper = messageWrapper(msg);
-            }
-            catch (IOException e) {
-                log.log(Level.SEVERE, "Failed to marshal message: " + msg, e);
-
-                removePending(reqId);
-
-                fut.onDone(e);
-
-                return fut;
-            }
-
-            GridNioFuture<?> sndFut = ses.send(wrapper);
+            GridNioFuture<?> sndFut = ses.send(msg);
 
             lastMsgSndTime = U.currentTimeMillis();
 
@@ -473,59 +458,38 @@ public class GridClientNioTcpConnection extends GridClientConnection {
      * Handles incoming response message. If this connection is closed this method would signal empty event
      * if there is no more pending requests.
      *
-     * @param req Incoming response data.
+     * @param res Incoming response data.
      */
     @SuppressWarnings({"unchecked", "TooBroadScope"})
-    void handleResponse(GridClientMessageWrapper req) {
+    void handleResponse(GridClientMessage res) {
         lastMsgRcvTime = U.currentTimeMillis();
 
-        TcpClientFuture fut = pendingReqs.get(req.requestId());
+        TcpClientFuture fut = pendingReqs.get(res.requestId());
 
         if (fut == null) {
             log.warning("Response for an unknown request is received, ignoring. " +
-                "[req=" + req + ", ses=" + ses + ']');
+                "[res=" + res + ", ses=" + ses + ']');
 
             return;
         }
 
         if (fut.forward()) {
-            GridRouterResponse msg = new GridRouterResponse(
-                req.messageArray(),
-                req.requestId(),
-                clientId,
-                req.destinationId());
-
-            removePending(msg.requestId());
-
-            fut.onDone(msg);
+            // TODO: IGNITE-61
+//            GridRouterResponse msg = new GridRouterResponse(
+//                res.messageArray(),
+//                res.requestId(),
+//                clientId,
+//                res.destinationId());
+//
+//            removePending(msg.requestId());
+//
+//            fut.onDone(msg);
         }
         else {
-            GridClientMessage msg;
-
-            if (keepPortablesMode != null)
-                keepPortablesMode.set(fut.keepPortables());
-
-            try {
-                msg = marsh.unmarshal(req.messageArray());
-            }
-            catch (IOException e) {
-                fut.onDone(new GridClientException("Failed to unmarshal message.", e));
-
-                return;
-            }
-
-            finally {
-                if (keepPortablesMode != null)
-                    keepPortablesMode.set(true);
-            }
-            msg.requestId(req.requestId());
-            msg.clientId(req.clientId());
-            msg.destinationId(req.destinationId());
-
-            if (msg instanceof GridClientResponse)
-                handleClientResponse(fut, (GridClientResponse)msg);
+            if (res instanceof GridClientResponse)
+                handleClientResponse(fut, (GridClientResponse)res);
             else
-                log.warning("Unsupported response type received: " + msg);
+                log.warning("Unsupported response type received: " + res);
         }
     }
 
@@ -559,22 +523,7 @@ public class GridClientNioTcpConnection extends GridClientConnection {
 
                     req.requestId(resp.requestId());
 
-                    GridClientMessageWrapper wrapper;
-
-                    try {
-                        wrapper = messageWrapper(req);
-                    }
-                    catch (IOException e) {
-                        log.log(Level.SEVERE, "Failed to marshal message: " + req, e);
-
-                        removePending(resp.requestId());
-
-                        fut.onDone(e);
-
-                        return;
-                    }
-
-                    ses.send(wrapper);
+                    ses.send(req);
 
                     return;
                 }
@@ -588,22 +537,7 @@ public class GridClientNioTcpConnection extends GridClientConnection {
 
                     src.sessionToken(sesTok);
 
-                    GridClientMessageWrapper wrapper;
-
-                    try {
-                        wrapper = messageWrapper(src);
-                    }
-                    catch (IOException e) {
-                        log.log(Level.SEVERE, "Failed to marshal message: " + src, e);
-
-                        removePending(resp.requestId());
-
-                        fut.onDone(e);
-
-                        return;
-                    }
-
-                    ses.send(wrapper);
+                    ses.send(src);
 
                     return;
                 }
@@ -624,27 +558,6 @@ public class GridClientNioTcpConnection extends GridClientConnection {
     }
 
     /**
-     * @param msg Client message.
-     * @return Message wrapper for direct marshalling.
-     * @throws IOException If failed to marshal message.
-     */
-    private GridClientMessageWrapper messageWrapper(GridClientMessage msg) throws IOException {
-        GridClientMessageWrapper wrapper = new GridClientMessageWrapper();
-
-        wrapper.requestId(msg.requestId());
-        wrapper.clientId(clientId);
-        wrapper.destinationId(msg.destinationId());
-
-        ByteBuffer data = (msg instanceof GridRouterRequest) ? ByteBuffer.wrap(((GridRouterRequest)msg).body()) :
-            marsh.marshal(msg, 0);
-
-        wrapper.message(data);
-        wrapper.messageSize(data.remaining() + 40);
-
-        return wrapper;
-    }
-
-    /**
      * Removes pending request and signals to {@link #closedLatch} if necessary.
      *
      * @param reqId Request Id.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 0634b1f..0f46e3d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -34,7 +34,6 @@ import org.apache.ignite.internal.processors.clock.*;
 import org.apache.ignite.internal.processors.continuous.*;
 import org.apache.ignite.internal.processors.dataload.*;
 import org.apache.ignite.internal.processors.fs.*;
-import org.apache.ignite.internal.processors.rest.client.message.*;
 import org.apache.ignite.internal.processors.rest.handlers.task.*;
 import org.apache.ignite.internal.processors.rest.protocols.tcp.*;
 import org.apache.ignite.internal.processors.streamer.*;
@@ -445,26 +444,6 @@ public class GridIoMessageFactory implements MessageFactory {
 
                 break;
 
-            case 72:
-                msg = new GridClientHandshakeRequestWrapper();
-
-                break;
-
-            case 73:
-                msg = new GridClientHandshakeResponseWrapper();
-
-                break;
-
-            case 74:
-                msg = new GridClientMessageWrapper();
-
-                break;
-
-            case 75:
-                msg = new GridClientPingPacketWrapper();
-
-                break;
-
             case 76:
                 msg = new GridTaskResultRequest();
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeRequestWrapper.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeRequestWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeRequestWrapper.java
deleted file mode 100644
index 71dfcf9..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeRequestWrapper.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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.ignite.internal.processors.rest.client.message;
-
-import org.apache.ignite.internal.direct.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-
-import java.nio.*;
-
-/**
- * Client handshake wrapper for direct marshalling.
- */
-public class GridClientHandshakeRequestWrapper extends MessageAdapter {
-    /** */
-    private static final long serialVersionUID = -5705048094821942662L;
-
-    /** Signal char. */
-    public static final byte HANDSHAKE_HEADER = (byte)0x91;
-
-    /** Stream. */
-    private final DirectByteBufferStream stream = new DirectByteBufferStream(null);
-
-    /** Handshake bytes. */
-    private byte[] bytes;
-
-    /**
-     *
-     */
-    public GridClientHandshakeRequestWrapper() {
-        // No-op.
-    }
-
-    /**
-     *
-     * @param req Handshake request.
-     */
-    public GridClientHandshakeRequestWrapper(GridClientHandshakeRequest req) {
-        bytes = req.rawBytes();
-    }
-
-    /**
-     * @return Handshake bytes.
-     */
-    public byte[] bytes() {
-        return bytes;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean writeTo(ByteBuffer buf) {
-        stream.setBuffer(buf);
-
-        if (!typeWritten) {
-            if (!buf.hasRemaining())
-                return false;
-
-            stream.writeByte(directType());
-
-            typeWritten = true;
-        }
-
-        stream.writeByteArray(bytes, 0, bytes.length);
-
-        return stream.lastFinished();
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean readFrom(ByteBuffer buf) {
-        stream.setBuffer(buf);
-
-        bytes = stream.readByteArray(GridClientHandshakeRequest.PACKET_SIZE);
-
-        return stream.lastFinished();
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte directType() {
-        return HANDSHAKE_HEADER;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors"})
-    @Override public MessageAdapter clone() {
-        GridClientHandshakeRequestWrapper _clone = new GridClientHandshakeRequestWrapper();
-
-        clone0(_clone);
-
-        return _clone;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void clone0(MessageAdapter _msg) {
-        GridClientHandshakeRequestWrapper _clone = (GridClientHandshakeRequestWrapper)_msg;
-
-        _clone.bytes = bytes;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridClientHandshakeRequestWrapper.class, this);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeResponseWrapper.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeResponseWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeResponseWrapper.java
deleted file mode 100644
index e598f66..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientHandshakeResponseWrapper.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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.ignite.internal.processors.rest.client.message;
-
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-
-import java.nio.*;
-
-/**
- * Client handshake wrapper for direct marshalling.
- */
-public class GridClientHandshakeResponseWrapper extends MessageAdapter {
-    /** */
-    private static final long serialVersionUID = -1529807975073967381L;
-
-    /** */
-    private byte code;
-
-    /**
-     *
-     */
-    public GridClientHandshakeResponseWrapper() {
-        // No-op.
-    }
-
-    /**
-     * @param code Response code.
-     */
-    public GridClientHandshakeResponseWrapper(byte code) {
-        this.code = code;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean writeTo(ByteBuffer buf) {
-        writer.setBuffer(buf);
-
-        if (!typeWritten) {
-            if (!writer.writeByte(null, directType()))
-                return false;
-
-            typeWritten = true;
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean readFrom(ByteBuffer buf) {
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte directType() {
-        return code;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors"})
-    @Override public MessageAdapter clone() {
-        GridClientHandshakeResponseWrapper _clone = new GridClientHandshakeResponseWrapper();
-
-        clone0(_clone);
-
-        return _clone;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void clone0(MessageAdapter _msg) {
-        GridClientHandshakeResponseWrapper _clone = (GridClientHandshakeResponseWrapper)_msg;
-
-        _clone.code = code;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridClientHandshakeResponseWrapper.class, this);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientMessageWrapper.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientMessageWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientMessageWrapper.java
deleted file mode 100644
index 746a40f..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientMessageWrapper.java
+++ /dev/null
@@ -1,275 +0,0 @@
-/*
- * 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.ignite.internal.processors.rest.client.message;
-
-import org.apache.ignite.internal.direct.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-
-import java.nio.*;
-import java.util.*;
-
-/**
- * Client message wrapper for direct marshalling.
- */
-public class GridClientMessageWrapper extends MessageAdapter {
-    /** */
-    private static final long serialVersionUID = 5284375300887454697L;
-
-    /** Client request header. */
-    public static final byte REQ_HEADER = (byte)0x90;
-
-    /** Stream. */
-    private final DirectByteBufferStream stream = new DirectByteBufferStream(null);
-
-    /** */
-    private int msgSize;
-
-    /** */
-    private long reqId;
-
-    /** */
-    private UUID clientId;
-
-    /** */
-    private UUID destId;
-
-    /** */
-    private ByteBuffer msg;
-
-    /**
-     * @return Request ID.
-     */
-    public long requestId() {
-        return reqId;
-    }
-
-    /**
-     * @param reqId Request ID.
-     */
-    public void requestId(long reqId) {
-        this.reqId = reqId;
-    }
-
-    /**
-     * @return Message size.
-     */
-    public int messageSize() {
-        return msgSize;
-    }
-
-    /**
-     * @param msgSize Message size.
-     */
-    public void messageSize(int msgSize) {
-        this.msgSize = msgSize;
-    }
-
-    /**
-     * @return Client ID.
-     */
-    public UUID clientId() {
-        return clientId;
-    }
-
-    /**
-     * @param clientId Client ID.
-     */
-    public void clientId(UUID clientId) {
-        this.clientId = clientId;
-    }
-
-    /**
-     * @return Destination ID.
-     */
-    public UUID destinationId() {
-        return destId;
-    }
-
-    /**
-     * @param destId Destination ID.
-     */
-    public void destinationId(UUID destId) {
-        this.destId = destId;
-    }
-
-    /**
-     * @return Message buffer.
-     */
-    public ByteBuffer message() {
-        return msg;
-    }
-
-    /**
-     * @return Message bytes.
-     */
-    public byte[] messageArray() {
-        assert msg.hasArray();
-        assert msg.position() == 0 && msg.remaining() == msg.capacity();
-
-        return msg.array();
-    }
-
-    /**
-     * @param msg Message bytes.
-     */
-    public void message(ByteBuffer msg) {
-        this.msg = msg;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean writeTo(ByteBuffer buf) {
-        stream.setBuffer(buf);
-
-        if (!typeWritten) {
-            if (stream.remaining() < 1)
-                return false;
-
-            stream.writeByte(directType());
-
-            typeWritten = true;
-        }
-
-        switch (state) {
-            case 0:
-                if (stream.remaining() < 4)
-                    return false;
-
-                stream.writeInt(msgSize);
-
-                state++;
-
-            case 1:
-                if (stream.remaining() < 8)
-                    return false;
-
-                stream.writeLong(reqId);
-
-                state++;
-
-            case 2:
-                if (stream.remaining() < 16)
-                    return false;
-
-                stream.writeByteArray(U.uuidToBytes(clientId), 0, 16);
-
-                state++;
-
-            case 3:
-                if (stream.remaining() < 16)
-                    return false;
-
-                stream.writeByteArray(U.uuidToBytes(destId), 0, 16);
-
-                state++;
-
-            case 4:
-                stream.writeByteArray(msg.array(), msg.position(), msg.remaining());
-
-                state++;
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean readFrom(ByteBuffer buf) {
-        stream.setBuffer(buf);
-
-        switch (state) {
-            case 0:
-                if (stream.remaining() < 4)
-                    return false;
-
-                msgSize = stream.readInt();
-
-                if (msgSize == 0) // Ping message.
-                    return true;
-
-                state++;
-
-            case 1:
-                if (stream.remaining() < 8)
-                    return false;
-
-                reqId = stream.readLong();
-
-                state++;
-
-            case 2:
-                if (stream.remaining() < 16)
-                    return false;
-
-                clientId = U.bytesToUuid(stream.readByteArray(16), 0);
-
-                state++;
-
-            case 3:
-                if (stream.remaining() < 16)
-                    return false;
-
-                destId = U.bytesToUuid(stream.readByteArray(16), 0);
-
-                state++;
-
-            case 4:
-                byte[] msg0 = stream.readByteArray(msgSize);
-
-                if (!stream.lastFinished())
-                    return false;
-
-                msg = ByteBuffer.wrap(msg0);
-
-                state++;
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte directType() {
-        return REQ_HEADER;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors"})
-    @Override public MessageAdapter clone() {
-        GridClientMessageWrapper _clone = new GridClientMessageWrapper();
-
-        clone0(_clone);
-
-        return _clone;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void clone0(MessageAdapter _msg) {
-        GridClientMessageWrapper _clone = (GridClientMessageWrapper)_msg;
-
-        _clone.reqId = reqId;
-        _clone.msgSize = msgSize;
-        _clone.clientId = clientId;
-        _clone.destId = destId;
-        _clone.msg = msg;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridClientMessageWrapper.class, this);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientPingPacketWrapper.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientPingPacketWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientPingPacketWrapper.java
deleted file mode 100644
index ca8e804..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/GridClientPingPacketWrapper.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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.ignite.internal.processors.rest.client.message;
-
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-
-import java.nio.*;
-
-/**
- * Ping packet wrapper for direct marshalling.
- */
-public class GridClientPingPacketWrapper extends MessageAdapter {
-    /** */
-    private static final long serialVersionUID = -3956036611004055629L;
-
-    /** Ping message size (always zero). */
-    private int size;
-
-    /** {@inheritDoc} */
-    @Override public boolean writeTo(ByteBuffer buf) {
-        writer.setBuffer(buf);
-
-        if (!typeWritten) {
-            if (!writer.writeByte(null, directType()))
-                return false;
-
-            typeWritten = true;
-        }
-
-        switch (state) {
-            case 0:
-                if (!writer.writeInt("size", size))
-                    return false;
-
-                state++;
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean readFrom(ByteBuffer buf) {
-        throw new UnsupportedOperationException();
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte directType() {
-        return GridClientMessageWrapper.REQ_HEADER;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors"})
-    @Override public MessageAdapter clone() {
-        GridClientPingPacketWrapper _clone = new GridClientPingPacketWrapper();
-
-        clone0(_clone);
-
-        return _clone;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void clone0(MessageAdapter _msg) {
-        GridClientPingPacketWrapper _clone = (GridClientPingPacketWrapper)_msg;
-
-        _clone.size = size;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridClientPingPacketWrapper.class, this);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridClientPacketType.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridClientPacketType.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridClientPacketType.java
index 0bc8099..e47e825 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridClientPacketType.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridClientPacketType.java
@@ -27,6 +27,9 @@ public enum GridClientPacketType {
     /** Ignite handshake. */
     IGNITE_HANDSHAKE,
 
+    /** Ignite handshake response. */
+    IGNITE_HANDSHAKE_RES,
+
     /** Ignite message. */
     IGNITE
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridMemcachedMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridMemcachedMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridMemcachedMessage.java
index 707383f..a42fb57 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridMemcachedMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridMemcachedMessage.java
@@ -51,6 +51,9 @@ public class GridMemcachedMessage implements GridClientMessage {
     /** Client handshake flag. */
     public static final byte IGNITE_HANDSHAKE_FLAG = (byte)0x91;
 
+    /** Client handshake flag. */
+    public static final byte IGNITE_HANDSHAKE_RES_FLAG = (byte)0x92;
+
     /** Success status. */
     public static final int SUCCESS = 0x0000;
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestDirectParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestDirectParser.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestDirectParser.java
deleted file mode 100644
index b743e66..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestDirectParser.java
+++ /dev/null
@@ -1,523 +0,0 @@
-/*
- * 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.ignite.internal.processors.rest.protocols.tcp;
-
-import org.apache.ignite.*;
-import org.apache.ignite.client.marshaller.*;
-import org.apache.ignite.internal.direct.*;
-import org.apache.ignite.internal.processors.rest.client.message.*;
-import org.apache.ignite.internal.util.nio.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.nio.*;
-import java.nio.charset.*;
-import java.util.*;
-
-import static org.apache.ignite.internal.processors.rest.protocols.tcp.GridMemcachedMessage.*;
-import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.*;
-
-/**
- *
- */
-public class GridTcpRestDirectParser implements GridNioParser {
-    /** UTF-8 charset. */
-    private static final Charset UTF_8 = Charset.forName("UTF-8");
-
-    /** Message metadata key. */
-    private static final int MSG_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
-
-    /** Message reader. */
-    private final MessageReader rdr = new DirectMessageReader(null);
-
-    /** Protocol handler. */
-    private final GridTcpRestProtocol proto;
-
-    /**
-     * @param proto Protocol handler.
-     */
-    public GridTcpRestDirectParser(GridTcpRestProtocol proto) {
-        this.proto = proto;
-    }
-
-    /** {@inheritDoc} */
-    @Nullable @Override public Object decode(GridNioSession ses, ByteBuffer buf)
-        throws IOException, IgniteCheckedException {
-        ParserState state = ses.removeMeta(PARSER_STATE.ordinal());
-
-        if (state != null) {
-            assert state.packetType() == GridClientPacketType.MEMCACHE;
-
-            Object memcacheMsg = parseMemcachePacket(ses, buf, state);
-
-            if (memcacheMsg == null)
-                ses.addMeta(PARSER_STATE.ordinal(), state);
-
-            return memcacheMsg;
-        }
-
-        MessageAdapter msg = ses.removeMeta(MSG_META_KEY);
-
-        if (msg == null && buf.hasRemaining()) {
-            byte type = buf.get(buf.position());
-
-            if (type == GridClientMessageWrapper.REQ_HEADER) {
-                buf.get();
-
-                msg = new GridClientMessageWrapper();
-
-                msg.setReader(rdr);
-            }
-            else if (type == GridClientHandshakeRequestWrapper.HANDSHAKE_HEADER) {
-                buf.get();
-
-                msg = new GridClientHandshakeRequestWrapper();
-
-                msg.setReader(rdr);
-            }
-            else if (type == MEMCACHE_REQ_FLAG) {
-                state = new ParserState();
-
-                state.packet(new GridMemcachedMessage());
-                state.packetType(GridClientPacketType.MEMCACHE);
-
-                Object memcacheMsg = parseMemcachePacket(ses, buf, state);
-
-                if (memcacheMsg == null)
-                    ses.addMeta(PARSER_STATE.ordinal(), state);
-
-                return memcacheMsg;
-            }
-            else
-                throw new IOException("Invalid message type: " + type);
-        }
-
-        boolean finished = false;
-
-        if (buf.hasRemaining())
-            finished = msg.readFrom(buf);
-
-        if (finished) {
-            if (msg instanceof GridClientMessageWrapper) {
-                GridClientMessageWrapper clientMsg = (GridClientMessageWrapper)msg;
-
-                if (clientMsg.messageSize() == 0)
-                    return GridClientPingPacket.PING_MESSAGE;
-
-                GridClientMarshaller marsh = proto.marshaller(ses);
-
-                GridClientMessage ret = marsh.unmarshal(clientMsg.messageArray());
-
-                ret.requestId(clientMsg.requestId());
-                ret.clientId(clientMsg.clientId());
-                ret.destinationId(clientMsg.destinationId());
-
-                return ret;
-            }
-            else {
-                assert msg instanceof GridClientHandshakeRequestWrapper;
-
-                GridClientHandshakeRequestWrapper req = (GridClientHandshakeRequestWrapper)msg;
-
-                GridClientHandshakeRequest ret = new GridClientHandshakeRequest();
-
-                ret.putBytes(req.bytes(), 0, 4);
-
-                return ret;
-            }
-        }
-        else {
-            ses.addMeta(MSG_META_KEY, msg);
-
-            return null;
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public ByteBuffer encode(GridNioSession ses, Object msg) throws IOException, IgniteCheckedException {
-        // No encoding needed for direct messages.
-        throw new UnsupportedEncodingException();
-    }
-
-    /**
-     * Parses memcache protocol message.
-     *
-     * @param ses Session.
-     * @param buf Buffer containing not parsed bytes.
-     * @param state Current parser state.
-     * @return Parsed packet.s
-     * @throws IOException If packet cannot be parsed.
-     * @throws IgniteCheckedException If deserialization error occurred.
-     */
-    @Nullable private GridClientMessage parseMemcachePacket(GridNioSession ses, ByteBuffer buf, ParserState state)
-        throws IOException, IgniteCheckedException {
-        assert state.packetType() == GridClientPacketType.MEMCACHE;
-        assert state.packet() != null;
-        assert state.packet() instanceof GridMemcachedMessage;
-
-        GridMemcachedMessage req = (GridMemcachedMessage)state.packet();
-        ByteArrayOutputStream tmp = state.buffer();
-        int i = state.index();
-
-        while (buf.remaining() > 0) {
-            byte b = buf.get();
-
-            if (i == 0)
-                req.requestFlag(b);
-            else if (i == 1)
-                req.operationCode(b);
-            else if (i == 2 || i == 3) {
-                tmp.write(b);
-
-                if (i == 3) {
-                    req.keyLength(U.bytesToShort(tmp.toByteArray(), 0));
-
-                    tmp.reset();
-                }
-            }
-            else if (i == 4)
-                req.extrasLength(b);
-            else if (i >= 8 && i <= 11) {
-                tmp.write(b);
-
-                if (i == 11) {
-                    req.totalLength(U.bytesToInt(tmp.toByteArray(), 0));
-
-                    tmp.reset();
-                }
-            }
-            else if (i >= 12 && i <= 15) {
-                tmp.write(b);
-
-                if (i == 15) {
-                    req.opaque(tmp.toByteArray());
-
-                    tmp.reset();
-                }
-            }
-            else if (i >= HDR_LEN && i < HDR_LEN + req.extrasLength()) {
-                tmp.write(b);
-
-                if (i == HDR_LEN + req.extrasLength() - 1) {
-                    req.extras(tmp.toByteArray());
-
-                    tmp.reset();
-                }
-            }
-            else if (i >= HDR_LEN + req.extrasLength() &&
-                i < HDR_LEN + req.extrasLength() + req.keyLength()) {
-                tmp.write(b);
-
-                if (i == HDR_LEN + req.extrasLength() + req.keyLength() - 1) {
-                    req.key(tmp.toByteArray());
-
-                    tmp.reset();
-                }
-            }
-            else if (i >= HDR_LEN + req.extrasLength() + req.keyLength() &&
-                i < HDR_LEN + req.totalLength()) {
-                tmp.write(b);
-
-                if (i == HDR_LEN + req.totalLength() - 1) {
-                    req.value(tmp.toByteArray());
-
-                    tmp.reset();
-                }
-            }
-
-            if (i == HDR_LEN + req.totalLength() - 1)
-                // Assembled the packet.
-                return assemble(ses, req);
-
-            i++;
-        }
-
-        state.index(i);
-
-        return null;
-    }
-
-    /**
-     * Validates incoming packet and deserializes all fields that need to be deserialized.
-     *
-     * @param ses Session on which packet is being parsed.
-     * @param req Raw packet.
-     * @return Same packet with fields deserialized.
-     * @throws IOException If parsing failed.
-     * @throws IgniteCheckedException If deserialization failed.
-     */
-    private GridClientMessage assemble(GridNioSession ses, GridMemcachedMessage req) throws IOException, IgniteCheckedException {
-        byte[] extras = req.extras();
-
-        // First, decode key and value, if any
-        if (req.key() != null || req.value() != null) {
-            short keyFlags = 0;
-            short valFlags = 0;
-
-            if (req.hasFlags()) {
-                if (extras == null || extras.length < FLAGS_LENGTH)
-                    throw new IOException("Failed to parse incoming packet (flags required for command) [ses=" +
-                        ses + ", opCode=" + Integer.toHexString(req.operationCode() & 0xFF) + ']');
-
-                keyFlags = U.bytesToShort(extras, 0);
-                valFlags = U.bytesToShort(extras, 2);
-            }
-
-            if (req.key() != null) {
-                assert req.key() instanceof byte[];
-
-                byte[] rawKey = (byte[])req.key();
-
-                // Only values can be hessian-encoded.
-                req.key(decodeObj(keyFlags, rawKey));
-            }
-
-            if (req.value() != null) {
-                assert req.value() instanceof byte[];
-
-                byte[] rawVal = (byte[])req.value();
-
-                req.value(decodeObj(valFlags, rawVal));
-            }
-        }
-
-        if (req.hasExpiration()) {
-            if (extras == null || extras.length < 8)
-                throw new IOException("Failed to parse incoming packet (expiration value required for command) [ses=" +
-                    ses + ", opCode=" + Integer.toHexString(req.operationCode() & 0xFF) + ']');
-
-            req.expiration(U.bytesToInt(extras, 4) & 0xFFFFFFFFL);
-        }
-
-        if (req.hasInitial()) {
-            if (extras == null || extras.length < 16)
-                throw new IOException("Failed to parse incoming packet (initial value required for command) [ses=" +
-                    ses + ", opCode=" + Integer.toHexString(req.operationCode() & 0xFF) + ']');
-
-            req.initial(U.bytesToLong(extras, 8));
-        }
-
-        if (req.hasDelta()) {
-            if (extras == null || extras.length < 8)
-                throw new IOException("Failed to parse incoming packet (delta value required for command) [ses=" +
-                    ses + ", opCode=" + Integer.toHexString(req.operationCode() & 0xFF) + ']');
-
-            req.delta(U.bytesToLong(extras, 0));
-        }
-
-        if (extras != null) {
-            // Clients that include cache name must always include flags.
-            int len = 4;
-
-            if (req.hasExpiration())
-                len += 4;
-
-            if (req.hasDelta())
-                len += 8;
-
-            if (req.hasInitial())
-                len += 8;
-
-            if (extras.length - len > 0) {
-                byte[] cacheName = new byte[extras.length - len];
-
-                U.arrayCopy(extras, len, cacheName, 0, extras.length - len);
-
-                req.cacheName(new String(cacheName, UTF_8));
-            }
-        }
-
-        return req;
-    }
-
-    /**
-     * Decodes value from a given byte array to the object according to the flags given.
-     *
-     * @param flags Flags.
-     * @param bytes Byte array to decode.
-     * @return Decoded value.
-     * @throws IgniteCheckedException If deserialization failed.
-     */
-    private Object decodeObj(short flags, byte[] bytes) throws IgniteCheckedException {
-        assert bytes != null;
-
-        if ((flags & SERIALIZED_FLAG) != 0)
-            return proto.jdkMarshaller().unmarshal(bytes, null);
-
-        int masked = flags & 0xff00;
-
-        switch (masked) {
-            case BOOLEAN_FLAG:
-                return bytes[0] == '1';
-            case INT_FLAG:
-                return U.bytesToInt(bytes, 0);
-            case LONG_FLAG:
-                return U.bytesToLong(bytes, 0);
-            case DATE_FLAG:
-                return new Date(U.bytesToLong(bytes, 0));
-            case BYTE_FLAG:
-                return bytes[0];
-            case FLOAT_FLAG:
-                return Float.intBitsToFloat(U.bytesToInt(bytes, 0));
-            case DOUBLE_FLAG:
-                return Double.longBitsToDouble(U.bytesToLong(bytes, 0));
-            case BYTE_ARR_FLAG:
-                return bytes;
-            default:
-                return new String(bytes, UTF_8);
-        }
-    }
-
-    /**
-     * Holder for parser state and temporary buffer.
-     */
-    protected static class ParserState {
-        /** Parser index. */
-        private int idx;
-
-        /** Temporary data buffer. */
-        private ByteArrayOutputStream buf = new ByteArrayOutputStream();
-
-        /** Packet being assembled. */
-        private GridClientMessage packet;
-
-        /** Packet type. */
-        private GridClientPacketType packetType;
-
-        /** Header data. */
-        private HeaderData hdr;
-
-        /**
-         * @return Stored parser index.
-         */
-        public int index() {
-            return idx;
-        }
-
-        /**
-         * @param idx Index to store.
-         */
-        public void index(int idx) {
-            this.idx = idx;
-        }
-
-        /**
-         * @return Temporary data buffer.
-         */
-        public ByteArrayOutputStream buffer() {
-            return buf;
-        }
-
-        /**
-         * @return Pending packet.
-         */
-        @Nullable public GridClientMessage packet() {
-            return packet;
-        }
-
-        /**
-         * @param packet Pending packet.
-         */
-        public void packet(GridClientMessage packet) {
-            assert this.packet == null;
-
-            this.packet = packet;
-        }
-
-        /**
-         * @return Pending packet type.
-         */
-        public GridClientPacketType packetType() {
-            return packetType;
-        }
-
-        /**
-         * @param packetType Pending packet type.
-         */
-        public void packetType(GridClientPacketType packetType) {
-            this.packetType = packetType;
-        }
-
-        /**
-         * @return Header.
-         */
-        public HeaderData header() {
-            return hdr;
-        }
-
-        /**
-         * @param hdr Header.
-         */
-        public void header(HeaderData hdr) {
-            this.hdr = hdr;
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(ParserState.class, this);
-        }
-    }
-
-    /**
-     * Header.
-     */
-    protected static class HeaderData {
-        /** Request Id. */
-        private final long reqId;
-
-        /** Request Id. */
-        private final UUID clientId;
-
-        /** Request Id. */
-        private final UUID destId;
-
-        /**
-         * @param reqId Request Id.
-         * @param clientId Client Id.
-         * @param destId Destination Id.
-         */
-        private HeaderData(long reqId, UUID clientId, UUID destId) {
-            this.reqId = reqId;
-            this.clientId = clientId;
-            this.destId = destId;
-        }
-
-        /**
-         * @return Request Id.
-         */
-        public long reqId() {
-            return reqId;
-        }
-
-        /**
-         * @return Client Id.
-         */
-        public UUID clientId() {
-            return clientId;
-        }
-
-        /**
-         * @return Destination Id.
-         */
-        public UUID destinationId() {
-            return destId;
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestNioListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestNioListener.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestNioListener.java
index 12dc711..159f178 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestNioListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestNioListener.java
@@ -29,14 +29,11 @@ import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.jetbrains.annotations.*;
 
-import java.io.*;
-import java.nio.*;
 import java.util.*;
 import java.util.concurrent.*;
 
 import static org.apache.ignite.internal.processors.rest.GridRestCommand.*;
 import static org.apache.ignite.internal.processors.rest.client.message.GridClientCacheRequest.GridCacheOperation.*;
-import static org.apache.ignite.internal.processors.rest.client.message.GridClientHandshakeResponse.*;
 import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.*;
 
 /**
@@ -138,7 +135,7 @@ public class GridTcpRestNioListener extends GridNioServerListenerAdapter<GridCli
             memcachedLsnr.onMessage(ses, (GridMemcachedMessage)msg);
         else {
             if (msg == GridClientPingPacket.PING_MESSAGE)
-                ses.send(new GridClientPingPacketWrapper());
+                ses.send(new GridClientPingPacket());
             else if (msg instanceof GridClientHandshakeRequest) {
                 GridClientHandshakeRequest hs = (GridClientHandshakeRequest)msg;
 
@@ -168,7 +165,7 @@ public class GridTcpRestNioListener extends GridNioServerListenerAdapter<GridCli
                     else {
                         ses.addMeta(MARSHALLER.ordinal(), marsh);
 
-                        ses.send(new GridClientHandshakeResponseWrapper(CODE_OK));
+                        ses.send(GridClientHandshakeResponse.OK);
                     }
                 }
             }
@@ -205,27 +202,7 @@ public class GridTcpRestNioListener extends GridNioServerListenerAdapter<GridCli
                                 res.errorMessage("Failed to process client request: " + e.getMessage());
                             }
 
-                            GridClientMessageWrapper wrapper = new GridClientMessageWrapper();
-
-                            wrapper.requestId(msg.requestId());
-                            wrapper.clientId(msg.clientId());
-
-                            try {
-                                ByteBuffer bytes = proto.marshaller(ses).marshal(res, 0);
-
-                                wrapper.message(bytes);
-
-                                wrapper.messageSize(bytes.remaining() + 40);
-                            }
-                            catch (IOException e) {
-                                U.error(log, "Failed to marshal response: " + res, e);
-
-                                ses.close();
-
-                                return;
-                            }
-
-                            ses.send(wrapper);
+                            ses.send(res);
                         }
                     });
                 else

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
index d94b598..978ae61 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
@@ -81,6 +81,13 @@ public class GridTcpRestParser implements GridNioParser {
 
                     break;
 
+                case IGNITE_HANDSHAKE_RES_FLAG:
+                    buf.get();
+
+                    state.packetType(GridClientPacketType.IGNITE_HANDSHAKE_RES);
+
+                    break;
+
                 default:
                     throw new IOException("Failed to parse incoming packet (invalid packet start) [ses=" + ses +
                         ", b=" + Integer.toHexString(hdr & 0xFF) + ']');
@@ -100,6 +107,12 @@ public class GridTcpRestParser implements GridNioParser {
 
                 break;
 
+            case IGNITE_HANDSHAKE_RES:
+                if (buf.hasRemaining())
+                    res = new GridClientHandshakeResponse(buf.get());
+
+                break;
+
             case IGNITE:
                 res = parseCustomPacket(ses, buf, state);
 
@@ -123,8 +136,21 @@ public class GridTcpRestParser implements GridNioParser {
             return encodeMemcache((GridMemcachedMessage)msg);
         else if (msg == GridClientPingPacket.PING_MESSAGE)
             return ByteBuffer.wrap(GridClientPingPacket.PING_PACKET);
+        else if (msg instanceof GridClientHandshakeRequest) {
+            byte[] bytes = ((GridClientHandshakeRequest)msg).rawBytes();
+
+            ByteBuffer buf = ByteBuffer.allocate(bytes.length + 1);
+
+            buf.put(IGNITE_HANDSHAKE_FLAG);
+            buf.put(bytes);
+
+            buf.flip();
+
+            return buf;
+        }
         else if (msg instanceof GridClientHandshakeResponse)
             return ByteBuffer.wrap(new byte[] {
+                IGNITE_HANDSHAKE_RES_FLAG,
                 ((GridClientHandshakeResponse)msg).resultCode()
             });
         else {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9cb1573/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
index c0c63a4..19971e1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
@@ -106,7 +106,7 @@ public class GridTcpRestProtocol extends GridRestProtocolAdapter {
 
         lsnr = new GridTcpRestNioListener(log, this, hnd, ctx);
 
-        GridNioParser parser = new GridTcpRestDirectParser(this);
+        GridNioParser parser = new GridTcpRestParser();
 
         try {
             host = resolveRestTcpHost(ctx.config());
@@ -207,14 +207,14 @@ public class GridTcpRestProtocol extends GridRestProtocolAdapter {
     private boolean startTcpServer(InetAddress hostAddr, int port, GridNioServerListener<GridClientMessage> lsnr,
         GridNioParser parser, @Nullable SSLContext sslCtx, ClientConnectionConfiguration cfg) {
         try {
-            GridNioFilter codec = new GridNioCodecFilter(parser, log, true);
+            GridNioFilter codec = new GridNioCodecFilter(parser, log, false);
 
             GridNioFilter[] filters;
 
             if (sslCtx != null) {
                 GridNioSslFilter sslFilter = new GridNioSslFilter(sslCtx, log);
 
-                sslFilter.directMode(true);
+                sslFilter.directMode(false);
 
                 boolean auth = cfg.isRestTcpSslClientAuth();
 
@@ -244,7 +244,7 @@ public class GridTcpRestProtocol extends GridRestProtocolAdapter {
                 .socketReceiveBufferSize(cfg.getRestTcpReceiveBufferSize())
                 .sendQueueLimit(cfg.getRestTcpSendQueueLimit())
                 .filters(filters)
-                .directMode(true)
+                .directMode(false)
                 .build();
 
             srv.idleTimeout(cfg.getRestIdleTimeout());


Mime
View raw message