<commit_before>﻿using System;
using System.Threading.Tasks;
using EasyNetQ.Topology;

namespace EasyNetQ.Rpc
{
    class BetterRpc : IRpc
    {
        private readonly IAdvancedClientRpc _clientRpc;
        private readonly IAdvancedServerRpc _serverRpc;
        private readonly IConventions _conventions;
        private readonly IMessageSerializationStrategy _messageSerializationStrategy;

        public BetterRpc(IAdvancedBus advancedBus, IAdvancedRpcFactory advancedRpcFactory, IConventions conventions, IMessageSerializationStrategy messageSerializationStrategy)
        {
            _clientRpc = advancedRpcFactory.CreateClientRpc(advancedBus);
            _serverRpc = advancedRpcFactory.CreateServerRpc(advancedBus);
            _conventions = conventions;
            _messageSerializationStrategy = messageSerializationStrategy;
        }

        public Task<TResponse> Request<TRequest, TResponse>(TRequest request)
            where TRequest : class
            where TResponse : class
        {
            var exchange = new Exchange(_conventions.RpcExchangeNamingConvention());
            var requestRoutingKey = _conventions.RpcRoutingKeyNamingConvention(typeof(TRequest));
            var message = new Message<TRequest>(request);

            var serializedMessage = _messageSerializationStrategy.SerializeMessage(message);

            var response = _clientRpc.RequestAsync(exchange, requestRoutingKey, false, false, () => _conventions.RpcReturnQueueNamingConvention(), serializedMessage);
            return response.ContinueWith(
                task => ((IMessage<TResponse>)_messageSerializationStrategy.DeserializeMessage(task.Result.Properties, task.Result.Body).Message).Body,
                TaskContinuationOptions.NotOnFaulted);
        }

        //TODO add handlerId
        public IDisposable Respond<TRequest, TResponse>(Func<TRequest, Task<TResponse>> responder)
            where TRequest : class
            where TResponse : class
        {
            var handlerId = "";

            var exchange = new Exchange(_conventions.RpcExchangeNamingConvention());
            var queue = new Queue(_conventions.RpcRequestQueueNameConvention(typeof (TRequest), handlerId),false);
            var topic = _conventions.RpcRoutingKeyNamingConvention(typeof(TRequest));

            return _serverRpc.Respond(exchange, queue, topic, HandleRequest(responder));
        }

        private Func<SerializedMessage, Task<SerializedMessage>> HandleRequest<TRequest,TResponse>(Func<TRequest, Task<TResponse>> handle) where TResponse : class
        {
            return serializedMessage =>
                {
                    var deserializedMessage = _messageSerializationStrategy.DeserializeMessage(serializedMessage.Properties, serializedMessage.Body);
                    var request = (IMessage<TRequest>) deserializedMessage.Message;
                    var responseTask = handle(request.Body);
                    return responseTask.ContinueWith(
                        task => _messageSerializationStrategy.SerializeMessage(new Message<TResponse>(task.Result)),
                        TaskContinuationOptions.NotOnFaulted);
                };
        }
    }
}<commit_msg>Fixed bug: Exception propagation from server to client.<commit_after>﻿using System;
using System.Threading.Tasks;
using EasyNetQ.Topology;

namespace EasyNetQ.Rpc
{
    class BetterRpc : IRpc
    {
        private readonly IAdvancedClientRpc _clientRpc;
        private readonly IAdvancedServerRpc _serverRpc;
        private readonly IConventions _conventions;
        private readonly IMessageSerializationStrategy _messageSerializationStrategy;

        public BetterRpc(IAdvancedBus advancedBus, IAdvancedRpcFactory advancedRpcFactory, IConventions conventions, IMessageSerializationStrategy messageSerializationStrategy)
        {
            _clientRpc = advancedRpcFactory.CreateClientRpc(advancedBus);
            _serverRpc = advancedRpcFactory.CreateServerRpc(advancedBus);
            _conventions = conventions;
            _messageSerializationStrategy = messageSerializationStrategy;
        }

        public Task<TResponse> Request<TRequest, TResponse>(TRequest request)
            where TRequest : class
            where TResponse : class
        {
            var exchange = new Exchange(_conventions.RpcExchangeNamingConvention());
            var requestRoutingKey = _conventions.RpcRoutingKeyNamingConvention(typeof(TRequest));
            var message = new Message<TRequest>(request);

            var serializedMessage = _messageSerializationStrategy.SerializeMessage(message);

            var response = _clientRpc.RequestAsync(exchange, requestRoutingKey, false, false, () => _conventions.RpcReturnQueueNamingConvention(), serializedMessage);
            return response.Then(sMsg => 
                TaskHelpers.FromResult(((IMessage<TResponse>) _messageSerializationStrategy.DeserializeMessage(sMsg.Properties, sMsg.Body).Message).Body));
        }

        //TODO add handlerId
        public IDisposable Respond<TRequest, TResponse>(Func<TRequest, Task<TResponse>> responder)
            where TRequest : class
            where TResponse : class
        {
            var handlerId = "";

            var exchange = new Exchange(_conventions.RpcExchangeNamingConvention());
            var queue = new Queue(_conventions.RpcRequestQueueNameConvention(typeof (TRequest), handlerId),false);
            var topic = _conventions.RpcRoutingKeyNamingConvention(typeof(TRequest));

            return _serverRpc.Respond(exchange, queue, topic, HandleRequest(responder));
        }

        private Func<SerializedMessage, Task<SerializedMessage>> HandleRequest<TRequest,TResponse>(Func<TRequest, Task<TResponse>> handle) where TResponse : class
        {
            return serializedMessage =>
                {
                    var deserializedMessage = _messageSerializationStrategy.DeserializeMessage(serializedMessage.Properties, serializedMessage.Body);
                    var request = (IMessage<TRequest>) deserializedMessage.Message;
                    var responseTask = handle(request.Body);
                    return responseTask.Then(response => 
                        TaskHelpers.FromResult(_messageSerializationStrategy.SerializeMessage(new Message<TResponse>(response))));
                };
        }

        
    }
}
{'fraction_non_alphanumeric': 0.0725250626566416, 'fraction_numerical': 0.0, 'mean_word_length': 2.9981214777708205, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4017878', 'n_tokens_mistral': 1541, 'n_tokens_neox': 1505, 'n_words': 296}