anchor.diff:  handle_cast(_Request, State) ->
     {noreply, State}.
   
  -%%% incomming TCP %%%%%%
  -handle_info({OK,Socket,<<127,L:4,S:4,0,0>>},
  -	  #state{ok=OK,socket=Socket,transport=Transport,enc=undefined,session=Session,erlbin_number=EBin}=State) ->
  -  % there is no special case for extra data comming along the line as this is against the spec.
  -  % the client waits for the reply ... if not then kill the line.
  -  MaxLength = round(math:pow(2,9+L)),
  -  {Enc,Reply} = case S of
  -                  0 ->
  -                    {illegal,<<127,0,0,0>>};
  -                  1 ->
  -                    {raw_json,<<127,?MAXLENGTH:4,S:4,0,0>>};
  -                  2 ->
  -                    {raw_msgpack,<<127,?MAXLENGTH:4,S:4,0,0>>};
  -                  EBin ->
  -                    {raw_erlbin,<<127,?MAXLENGTH:4,S:4,0,0>>};
  -                  _ ->
  -                    {undefined,<<127,0,0,0>>}
  -                end,
  -
  -  {ok,Peer} = get_peer(Socket),
  -  Transport:send(Socket,Reply),
  -  case Enc of
  -    illegal -> {stop,normal,State};
  -    undefined -> {stop,normal,State};
  -    _ ->
  -      Transport:setopts(Socket, [{active, once}]),
  -      Session1 = erwa_session:set_source(tcp,Session),
  -      Session2 = erwa_session:set_peer(Peer,Session1),
  -      {noreply,State#state{enc=Enc,length=MaxLength,session=Session2}}
  -  end;
  -handle_info({OK,Socket,<<127,_L:4,_S:4,_,_>>},
  -	    #state{ok=OK,socket=Socket,transport=Transport,enc=undefined}=State) ->
  -  % stop any connection that is not sending the last two zeros.
  -  Transport:send(Socket,<<127,0,0,0>>),
  -  {stop,normal,State};
  -handle_info({OK,Socket,_},
  -        #state{ok=OK,socket=Socket,enc=undefined}=State) ->
  -  % just close any misbehaving client
  -  {stop,normal,State};
  -handle_info({OK,Socket,Data},
  -            #state{ok=OK,socket=Socket,transport=Transport,enc=Enc,buffer=Buffer,session=Session}=State) ->
  -  Transport:setopts(Socket, [{active, once}]),
  -  {Messages,NewBuffer} = wamper_protocol:deserialize(<<Buffer/binary, Data/binary>>,Enc),
  -  case handle_messages(Messages,Session,Transport,Socket,Enc) of
  -    {ok, NewSession} ->
  -      NewState = State#state{session=NewSession, buffer=NewBuffer},
  -      {noreply, NewState};
  -    {stop, NewSession} ->
  -      NewState = State#state{session=NewSession, buffer=NewBuffer},
  -      {stop,normal,NewState}
  -  end;
  -handle_info({Closed,Socket}, #state{closed=Closed,socket=Socket}=State) ->
  -  %wamper_protocol:close(connection_closed,ProtState),
  -  {stop, normal, State};
  -handle_info({Error,Socket,Reason}, #state{error=Error,socket=Socket}=State) ->
  -  {stop, {error, Reason} , State};
  -
  -
  -
  -%%% messages comming from routing / system (internal) %%%
  -
  -handle_info({erwa,Msg}, #state{socket=Socket,transport=Transport,enc=Enc,session=Session}=State) when is_tuple(Msg) ->
  -  case erwa_session:handle_info(Msg, Session) of
  -    {ok, NewSession} ->
  -      {noreply,State#state{session=NewSession}};
  -    {send, OutMsg, NewSession} ->
  -      Transport:send(Socket,wamper_protocol:serialize(OutMsg,Enc)),
  -      {noreply,State#state{session=NewSession}};
  -    {send_stop, OutMsg, NewSession} ->
  -      Transport:send(Socket,wamper_protocol:serialize(OutMsg,Enc)),
  -      {stop, normal, State#state{session=NewSession}};
  -    {stop, NewSession} ->
  -      {stop, normal, State#state{session=NewSession}}
  -  end;
  -handle_info(Info, State) ->
  -  erlang:error("tcp_in: unknown info message: ~p~n",[Info]),
  -  {noreply, State}.
  +
  +handle_info({Ok, Socket, Data}, #state{ok=Ok, socket=Socket}=State) ->
  +	handle_incomming_tcp(Data,State);
  +handle_info({Closed, Socket}, #state{closed=Closed, socket=Socket}=State) ->
  +	handle_socket_closed(State);
  +handle_info({Err,Socket,Error}, #state{error=Err,socket=Socket}=State) ->
  +	handle_socket_error(Error,State);
  +handle_info({erwa,Msg},State) ->
  +	handle_message_from_routing(Msg,State);
  +handle_info(Msg,State) ->
  +	handle_unsupported_info(Msg,State).
  +
  +
  +
  +
  +ack_otp_starting(Ref) ->
  +	ok = proc_lib:init_ack({ok, self()}),
  +	ok = ranch:accept_ack(Ref),
  +	ok.
  +
  +create_intial_state(Transport,Socket) ->
  +  {Ok,Closed, Error} = Transport:messages(),
  +  #state{
  +	 socket=Socket,
  +	 transport=Transport,
  +     ok=Ok,
  +	 closed=Closed,
  +	 error=Error,
  +	 session=erwa_session:create()
  +	 }.
  +
  +
  +handle_incomming_tcp(Data,#state{enc=undefined} = State ) ->
  +	handle_handshake(Data,State);
  +handle_incomming_tcp(Data, State) ->
  +	handle_incomming_wamp(Data,State).
  +
  +
  +handle_incomming_wamp(Data, #state{transport=Transport, socket=Socket}=State) ->
  +	enable_socket_once(Transport,Socket),
  +	deserialize_and_handle_incomming_messages(Data,State).
  +	
  +
  +deserialize_and_handle_incomming_messages(Data,State) ->
  +	{Messages, State} = deserialize_messages_update_state(Data,State),
  +	handle_incomming_wamp_messages( Messages, State ).
  +
  +deserialize_messages_update_state(Data,#state{buffer=Buffer, enc=Enc} = State) ->
  +	{Messages, NewBuffer} = wamper_protocol:deserialize(<<Buffer/binary,
  +														  Data/binary>>,Enc),
  +	{Messages, State#state{buffer=NewBuffer}}.
  +
  +
  +handle_incomming_wamp_messages([],#state{} = State) ->
  +	{noreply, State};
  +handle_incomming_wamp_messages([Msg | Tail], State) ->
  +	Result = handle_message_by_session(Msg,State),	
  +	handle_result_of_session(Result,Tail,State).
  +
  +
  +handle_result_of_session({ok,NewSession},Tail,State) ->
  +	handle_incomming_wamp_messages(Tail,State#state{session=NewSession});
  +handle_result_of_session({reply, OutMsg, NewSession},Tail,State) ->
  +	serialize_and_send_to_peer(OutMsg,State),
  +	handle_incomming_wamp_messages(Tail,State#state{session=NewSession});
  +handle_result_of_session({reply_stop,OutMsg, NewSession}, _Tail, State) ->
  +	serialize_and_send_to_peer(OutMsg, State),
  +	close_connection(State#state{session=NewSession});
  +handle_result_of_session({stop,NewSession},_Tail,State) ->
  +	close_connection(State#state{session=NewSession}).
  +
  +
  +
  +
  +handle_message_by_session(Msg,#state{session=Session}) ->
  +	erwa_session:handle_message(Msg,Session).
  +
  +
  +handle_message_from_routing(Msg,#state{session=Session}=State) ->
  +	Result = handle_routing_message_by_session(Msg,Session),
  +	handle_result_of_session_for_routing(Result,State).
  +
  +
  +handle_routing_message_by_session(Msg,Session) ->
  +	erwa_session:handle_info(Msg,Session).
  +
  +
  +handle_result_of_session_for_routing({ok, NewSession},State) ->
  +	{noreply,State#state{session=NewSession}};
  +handle_result_of_session_for_routing({send, OutMsg, NewSession},State) ->
  +	serialize_and_send_to_peer(OutMsg,State),
  +	{noreply,State#state{session=NewSession}};
  +handle_result_of_session_for_routing({send_stop, OutMsg, NewSession},State) ->
  +	serialize_and_send_to_peer(OutMsg,State),
  +	close_connection( State#state{session=NewSession});
  +handle_result_of_session_for_routing({stop, NewSession},State) ->
  +	close_connection(State#state{session=NewSession}).
  +
  +
  +handle_handshake(<<127,MaxLengthExp:4,ProtocolNumber:4,0,0>>, State) ->
  +	handle_handshake_message(MaxLengthExp,ProtocolNumber,State);
  +handle_handshake(<<127,_:4,_:4,_,_>>, State) ->
  +	handle_bad_handshake(State);
  +handle_handshake(_,State) ->
  +	handle_misbehaving_client(State).
  +
  +
  +handle_handshake_message(MaxLengthExp, ProtocolNumber, State ) ->
  +	ProtocolName = translate_protocol_number_to_name(ProtocolNumber),
  +	MaxLength = calculateMaxLength(MaxLengthExp),
  +	send_handshake_reply(ProtocolName, State),
  +	NewState = update_erwa_session_and_state(ProtocolName,MaxLength,State),
  +	closeOrActivateConnection(ProtocolName,NewState).
  +
  +
  +handle_bad_handshake(State) ->
  +	send_to_peer(unsupported_message(),State),
  +	close_connection(State).
  +
  +handle_misbehaving_client(State) ->
  +	close_connection(State).
  +
  +translate_protocol_number_to_name(0) -> unsupported;
  +translate_protocol_number_to_name(1) -> raw_json;
  +translate_protocol_number_to_name(2) -> raw_msgpack;
  +translate_protocol_number_to_name(N) ->
  +	ErlBin = erlbin_protocol_number(),
  +	if 
  +		N == ErlBin -> raw_erlbin;
  +		true -> unsupported
  +	end.
  +
  +
  +calculateMaxLength(MaxLengthExp) ->
  +	round(math:pow(2,9+MaxLengthExp)).
  +
  +send_handshake_reply(raw_json,State) ->
  +	send_to_peer(use_raw_json_message(), State);
  +send_handshake_reply(raw_msgpack, State) ->
  +	send_to_peer(use_raw_msgpack_message(), State);
  +send_handshake_reply(raw_erlbin, State) ->
  +	send_to_peer(use_raw_erlbin_message(), State);
  +send_handshake_reply(_, State) ->
  +	send_to_peer(unsupported_message(), State).
  +
  +update_erwa_session_and_state(unsupported,_,State) ->
  +	State;
  +update_erwa_session_and_state(ProtocolName, MaxLength, #state{session=Session,
  +														  transport=Transport
  +														 } = State) ->
  +	updateState(ProtocolName,
  +				MaxLength,
  +				updateErwaSession(Transport, 
  +								  get_peername(State), 
  +								  Session),
  +				State).
  +	
  +
  +updateState(ProtocolName, MaxLength, Session, State) ->
  +	State#state{
  +	  enc = ProtocolName,
  +	  length = MaxLength,
  +	  session = Session
  +	 }.
  +
  +updateErwaSession(Transport, Peer, Session) ->
  +	Session1 = erwa_session:set_peer(Peer,Session),
  +	Session2 = erwa_session:set_source(Transport,Session1),
  +	Session2.
  +
  +closeOrActivateConnection(unsupported,State) ->
  +	close_connection(State);
  +closeOrActivateConnection(_, State) ->
  + 	activateConnectionOnce(State).
  +
  +close_connection( State) ->
  +	{stop, normal, State}.
  +
  +activateConnectionOnce(#state{transport=Transport, socket=Socket} = State) ->
  +	ok = enable_socket_once(Transport,Socket),
  +	{noreply,State}.
  +
  +handle_unsupported_info(_Msg,#state{} = State) ->
  +	% @TODO: should be logged 
  +	{noreply, State}.
  +
  +enable_socket_once(Transport, Socket) ->
  +  ok = Transport:setopts(Socket, [{active, once}]),
  +  ok.
  +
  +handle_socket_closed(State) ->
  +	% need to close the session - maybe
  +	{stop, normal, State}.
  +
  +handle_socket_error(Error, State) ->
  +	% need to close the session - maybe
  +	{stop, {error, Error}, State }.
  +
  +
  +serialize_and_send_to_peer(Msg,#state{enc=Enc}=State) ->
  +	OutMsg = wamper_protocol:serialize(Msg,Enc),
  +	send_to_peer(OutMsg,State).
  +
  +
  +send_to_peer(Msg, #state{transport=Transport, socket=Socket}) ->
  +	Transport:send(Socket,Msg).
  +
  +
  +get_peername(#state{socket=Socket}) ->
  +	{ok, Peer} = inet:peername(Socket),
  +	Peer.
  +
  +
  +unsupported_message() ->
  +	<<127,0,0,0>>.
  +
  +use_raw_msgpack_message() ->
  +	<<127,?MAXLENGTH:4,2:4,0,0>>.
  +
  +use_raw_json_message() ->
  +	<<127,?MAXLENGTH:4,1:4,0,0>>.
  +
  +use_raw_erlbin_message() ->
  +	S = erlbin_protocol_number(),
  +	<<127,?MAXLENGTH:4,S:4,0,0>>.
   
   
   terminate(_Reason, _State) ->
  
positive.code:   {ok, State}.
  
  
  
  handle_messages([],Session,_Transport,_Socket,_Enc) ->
    {ok,Session};
  handle_messages([Msg|Tail],Session,Transport,Socket,Enc) ->
    case erwa_session:handle_message(Msg, Session) of
      {ok, NewSession} ->
        handle_messages(Tail,NewSession,Transport,Socket,Enc);
      {reply, OutMsg, NewSession} ->
        Transport:send(Socket,wamper_protocol:serialize(OutMsg,Enc)),
        handle_messages(Tail,NewSession,Transport,Socket,Enc);
      {reply_stop, OutMsg, NewSession} ->
        Transport:send(Socket,wamper_protocol:serialize(OutMsg,Enc)),
        {stop, NewSession};
      {stop, NewSession} ->
        {stop, NewSession}
    end.
  
  
  -ifndef(TEST).
  
  get_peer(Socket) ->
    inet:peername(Socket).
  
  
  -else.
  
  %% Test mocks
  
  get_peer(_) ->
    {ok,dummy}.
  
  %% tests
  
  handshake_hello_test() ->
    {ok,Socket} = erwa_tcp_mock:start(),
    Session = erwa_session:create(),
    State1 = #state{ok=ok,transport=erwa_tcp_mock,socket=Socket,closed=closed,error=error,session=Session},
  
    {stop,normal,_} = handle_info({ok,Socket,<<127,1:4,0:4,0,0>>},State1), %illegal
    {ok,<<127,0,0,0>>} = erwa_tcp_mock:get_buffer(Socket),
    {stop,normal,_} = handle_info({ok,Socket,<<127,1:4,13:4,0,0>>},State1), %undefined
    {ok,<<127,0,0,0>>} = erwa_tcp_mock:get_buffer(Socket),
    {stop,normal,_} = handle_info({ok,Socket,<<127,1:4,1:4,1,0>>},State1), %very bad
    {ok,<<127,0,0,0>>} = erwa_tcp_mock:get_buffer(Socket),
    {stop,normal,_} = handle_info({ok,Socket,<<128>>},State1), %very bad
    {ok,<<>>} = erwa_tcp_mock:get_buffer(Socket),
  
    {noreply,#state{length=Len1,enc=Enc1}} = handle_info({ok,Socket,<<127,1:4,1:4,0,0>>},State1),
    Len1 = 1024,
    Enc1 = raw_json,
    {ok,<<127,15:4,1:4,0,0>>} = erwa_tcp_mock:get_buffer(Socket),
  
    {noreply,#state{length=Len2,enc=Enc2}} = handle_info({ok,Socket,<<127,5:4,2:4,0,0>>},State1),
    Len2 = 16384,
    Enc2 = raw_msgpack,
    {ok,<<127,15:4,2:4,0,0>>} = erwa_tcp_mock:get_buffer(Socket),
  
    erwa_tcp_mock:stop(Socket).
  
  
  
  -endif.
  
negative.code:                ok,
                 closed,
                 error,
                 enc = undefined,
                 length = infitity,
                 buffer = <<"">>,
                 session = undefined
                }).
  
  
  % (x+9) ** 2 is the lengh
  % so
  %   0 -> 2 ** 9
  %   1 -> 2 ** 10 = 1024 etc.
  % the number gets shifted 4 bits to the left.
  % 15 is the max receive length possible ~ 16M (2 ** 24).
  % @TODO: make it configurable
  -define(MAXLENGTH, 15).
  
  %%% for TCP
  
  start_link(Ref, Socket, Transport, Opts) ->
  	start_link_tcp_connection_server(Ref, Socket, Transport, Opts).
  
  init(Ref, Socket, Transport, _Opts = []) ->
  	ok = ack_otp_starting(Ref),
  	ok = enable_socket_once(Transport, Socket),
  	State = create_intial_state(Transport,Socket),
  	gen_server:enter_loop(?MODULE, [], State).
  
  
  init(_Opts) ->
    erlang:error("don't call").
  