message cutting

Hi all,
i would like to ask for help.
sometimes i got message which is incomplete, zeroes on the end of the buffer.
Example:
System.FormatException: Deserialization of :[eyJDb3JyZWxhdGlvbklkIjoiMzM1MWNjNTktMzBmNi00NTk1LWFjYmQtMGRmYTZlMjIyYTRlIiwiRXhwaXJhdGlvblRpbWUiOm51bGwsIlNvdXJjZUFkZHJlc3MiOm51bGwsIlNlbnRUaW1lIjoiMjAxOS0xMS0wN1QxNTowOTowNC43NjQ4MzkzKzAxOjAwIiwiUGF5bG9hZFR5cGUiOiJLaXN0bGVyLkNkdC5Db250cmFjdHMuTG9nZ2VyLk1vZGVscy5JTG9nTWVzc2FnZSIsIlBheWxvYWQiOiJleUpNWlhabGJDSTZNaXdpVTI5MWNtTmxJam9pUzJsemRHeGxjaTVEWkhRdVdHMXNVbkJqVFdWemMyRm5aVUZrWVhCMFpYSk5hV1JrYkdWM1lYSmxMazFwWkdSc1pYZGhjbVV1VW1WeGRXVnpkRkpsYzNCdmJuTmxURzluWjJsdVowMXBaR1JzWlhkaGNtVWlMQ0pOWlhOellXZGxJam9pU0hSMGNDQlNaWE53YjI1elpTQkpibVp2Y20xaGRHbHZianBjWEhKY1hHNVRZMmhsYldFNmFIUjBjQ0JJYjNOME9pQXhNamN1TUM0d0xqRTZPREkzTlNCUVlYUm9PaUJjWEM5U1VFTXlJRkYxWlhKNVUzUnlhVzVuT2lBZ1VtVnpjRzl1YzJVZ1FtOWtlVG9nUEQ5NGJXd2dkbVZ5YzJsdmJqMWNYRndpTVM0d1hGeGNJaUJsYm1OdlpHbHVaejFjWEZ3aWRYUm1MVGhjWEZ3aVB6NWNYSEpjWEc0OGJXVjBhRzlrVW1WemNHOXVjMlUrWEZ4eVhGeHVJQ0E4Y0dGeVlXMXpQbHhjY2x4Y2JpQWdJQ0E4Y0dGeVlXMCtYRnh5WEZ4dUlDQWdJQ0FnUEhaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ1BHRnljbUY1UGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0E4WkdGMFlUNWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThkbUZzZFdVK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ0lDQThjM1J5YVc1blBuTjVjM1JsYlM1c2FYTjBUV1YwYUc5a2N6eGNYQzl6ZEhKcGJtYytYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQRnhjTDNaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJRHgyWVd4MVpUNWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQWdJRHh6ZEhKcGJtYytjM2x6ZEdWdExtMWxkR2h2WkZOcFoyNWhkSFZ5WlR4Y1hDOXpkSEpwYm1jK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ1BGeGNMM1poYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lEeDJZV3gxWlQ1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBZ0lEeHpkSEpwYm1jK2MzbHpkR1Z0TG0xbGRHaHZaRWhsYkhBOFhGd3ZjM1J5YVc1blBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lEeGNYQzkyWVd4MVpUNWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThkbUZzZFdVK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ0lDQThjM1J5YVc1blBtZGxkRjlyWlhsM2IzSmtYMjVoYldWelBGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NXlkVzVmYTJWNWQyOXlaRHhjWEM5emRISnBibWMrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEZ4Y0wzWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0FnSUR4emRISnBibWMrWjJWMFgydGxlWGR2Y21SZllYSm5kVzFsYm5SelBGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NW5aWFJmYTJWNWQyOXlaRjlrYjJOMWJXVnVkR0YwYVc5dVBGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NXpaWEoyYVdObExrZGxkRXhwYzNSUFpreHZZV1JsWkVWeGRXbHdiV1Z1ZER4Y1hDOXpkSEpwYm1jK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ1BGeGNMM1poYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lEeDJZV3gxWlQ1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBZ0lEeHpkSEpwYm1jK1kyRnNMbE5sZEUxdmJtbDBiM0pOYjJSbFBGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NWpZV3d1UjJWMFRXOXVhWFJ2Y2tacFpXeGtjenhjWEM5emRISnBibWMrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEZ4Y0wzWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0FnSUR4emRISnBibWMrWTJGc0xsTmxkRU52Ym5ScGJuVnZkWE5QZFhSd2RYUThYRnd2YzNSeWFXNW5QbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJRHhjWEM5MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0E4ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdJQ0E4YzNSeWFXNW5QbU5oYkM1VFpYUlFkV3h6WldSUGRYUndkWFE4WEZ3dmMzUnlhVzVuUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4Y1hDOTJZV3gxWlQ1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBOGRtRnNkV1UrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnSUNBOGMzUnlhVzVuUG1OaGJDNVRaWFJUYVc1MWMyOXBaR0ZzVDNWMGNIVjBQRnhjTDNOMGNtbHVaejVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0E4WEZ3dmRtRnNkV1UrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEhaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJQ0FnUEhOMGNtbHVaejVqWVd3dVUzZHBkR05vVDNWMGNIVjBRMmhoYm01bGJFOXVQRnhjTDNOMGNtbHVaejVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0E4WEZ3dmRtRnNkV1UrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEhaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJQ0FnUEhOMGNtbHVaejVqWVd3dVUzZHBkR05vVDNWMGNIVjBRMmhoYm01bGJFOW1aanhjWEM5emRISnBibWMrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEZ4Y0wzWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0FnSUR4emRISnBibWMrWkdGeExrZGxibVZ5WVhSbFJHTldiMngwWVdkbFBGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NWtZWEV1VFdWaGMzVnlaVlp2YkhSaFoyVkVZenhjWEM5emRISnBibWMrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEZ4Y0wzWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0FnSUR4emRISnBibWMrWkdGeExrMWxZWE4xY21WTmFXNU5ZWGc4WEZ3dmMzUnlhVzVuUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4Y1hDOTJZV3gxWlQ1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBOGRtRnNkV1UrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnSUNBOGMzUnlhVzVuUG1SaGNTNU5aV0Z6ZFhKbFZtOXNkR0ZuWlZCbFlXdFFaV0ZyUEZ4Y0wzTjBjbWx1Wno1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBOFhGd3ZkbUZzZFdVK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ1BIWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUNBZ1BITjBjbWx1Wno1a1lYRXVVbVZ6WlhSRVpYWnBZMlU4WEZ3dmMzUnlhVzVuUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4Y1hDOTJZV3gxWlQ1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBOGRtRnNkV1UrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnSUNBOGMzUnlhVzVuUG1SaGNTNVNaV0ZrUkdsbmFYUmhiRWx1Y0hWMFVHOXlkRHhjWEM5emRISnBibWMrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEZ4Y0wzWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUR4MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0FnSUR4emRISnBibWMrWkdGeExsTmxkRVJwWjJsMFlXeFBkWFJ3ZFhSTWFXNWxVM1JoZEdVOFhGd3ZjM1J5YVc1blBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lEeGNYQzkyWVd4MVpUNWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThkbUZzZFdVK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ0lDQThjM1J5YVc1blBtUmhjUzVIWlhSVGFXZHVZV3hUWVcxd2JHVnpQRnhjTDNOMGNtbHVaejVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0E4WEZ3dmRtRnNkV1UrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEhaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJQ0FnUEhOMGNtbHVaejVrYlcwdVVtVmhaRlpoYkhWbFBGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NWtiVzB1VW1WelpYUlViMFJsWm1GMWJIUlRaWFIwYVc1bmN6eGNYQzl6ZEhKcGJtYytYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQRnhjTDNaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJRHgyWVd4MVpUNWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQWdJRHh6ZEhKcGJtYytaRzF0TGxObGRFbHVjSFYwUEZ4Y0wzTjBjbWx1Wno1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBOFhGd3ZkbUZzZFdVK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ1BIWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUNBZ1BITjBjbWx1Wno1bVp5NVRkMmwwWTJoRGFHRnVibVZzVDI0OFhGd3ZjM1J5YVc1blBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lEeGNYQzkyWVd4MVpUNWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThkbUZzZFdVK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ0lDQThjM1J5YVc1blBtWm5MbE4zYVhSamFFTm9ZVzV1Wld4UFptWThYRnd2YzNSeWFXNW5QbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJRHhjWEM5MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0E4ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdJQ0E4YzNSeWFXNW5QbVpuTGxObGRFOTFkSEIxZER4Y1hDOXpkSEpwYm1jK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ1BGeGNMM1poYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lEeDJZV3gxWlQ1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBZ0lEeHpkSEpwYm1jK1ptY3VVMlYwUVhKaWFYUnlZWEo1VTJsbmJtRnNQRnhjTDNOMGNtbHVaejVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0E4WEZ3dmRtRnNkV1UrWEZ4eVhGeHVJQ0FnSUNBZ0lDQWdJQ0FnUEhaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJQ0FnUEhOMGNtbHVaejVtWnk1TWIyRmtRWEppYVhSeVlYSjVVMmxuYm1Gc1BGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NXZjMk11VW1WelpYUlViMFJsWm1GMWJIUlRaWFIwYVc1bmN6eGNYQzl6ZEhKcGJtYytYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQRnhjTDNaaGJIVmxQbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJRHgyWVd4MVpUNWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQWdJRHh6ZEhKcGJtYytiM05qTGxOMGIzQThYRnd2YzNSeWFXNW5QbHhjY2x4Y2JpQWdJQ0FnSUNBZ0lDQWdJRHhjWEM5MllXeDFaVDVjWEhKY1hHNGdJQ0FnSUNBZ0lDQWdJQ0E4ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdJQ0E4YzNSeWFXNW5QbTl6WXk1VFpYUlNkVzVOYjJSbFBGeGNMM04wY21sdVp6NWNYSEpjWEc0Z0lDQWdJQ0FnSUNBZ0lDQThYRnd2ZG1Gc2RXVStYRnh5WEZ4dUlDQWdJQ0FnSUNBZ0lDQWdQSFpoYkhWbFBseGNjbHhjYmlBZ0lDQWdJQ0FnSUNBZ0lDQWdQSE4wY21sdVp6NXZjMk11VjJGcGRGUnBiR3hCWTNGMWFYTnBkR2x2YmtOdmJYQnNaWFJsUEZ4Y0wzTjBjbWx1Wno1Y1hISmNYRzRnSUNBZ0lDQWdJQ0FnSUNBOFhGd3ZkbUZzZFdVK1hGeHlYRnh1SUNBZ0lDQWdJQ0FnSUNBZ1BIWmhiSFZsUGx4Y2NseGNiaUFnSUNBZ0lDQWdJQ0FnSUNBZ1BITjBjbWx1Wno1dmMyTXVWMkZwZEZWdWRHbHNWSEpwWjJkbGNrRnliV1ZrUEZ4Y0wzTjBjbWx1Wn ] failed ---> System.FormatException: The input is not a valid Base-64 string as it contains a non-base 64 character, more than two padding characters, or an illegal character among the padding characters.

white space on the end of message is null characters.

i have situation when creating subscribers and publishers just when need it, they are wrapped in using.

I am sending a json message which is encoded into base64 string.

my serializer methods:

public string SerializeToString(object _object)
{
return Base64Encode(JsonConvert.SerializeObject(_object));
}

    public object Deserialize(byte[] data, Type targetType)
    {
        try
        {
            var json = Base64Decode(_encoding.GetString(data));
            return JsonConvert.DeserializeObject(json, targetType);
        }
        catch (Exception e)
        {
            throw new FormatException($"Deserialization of :[{_encoding.GetString(data)}] failed", e);
        }
    }

i am using xpub-sub pattern. I have router with logging:

public class PubSubMessageRouter : IPubSubMessageRouter
{
    private Proxy _proxy;
    private NetMQPoller _poller;
    private XSubscriberSocket _xSub;
    private XPublisherSocket _xPub;
    private PushSocket _controlInPush = new PushSocket();
    private PushSocket _controlOutPush = new PushSocket();
    private PullSocket _controlInPull = new PullSocket();
    private PullSocket _controlOutPull = new PullSocket();
    private Task _workerReceived;
    private Task _workerSent;
    private CancellationTokenSource _cancellationTokenSource;
    private CancellationToken _cancellationToken;
    private WeakAction<MessageBase> _subscriberReceived;
    private WeakAction<MessageBase> _subscriberSent;
    private WeakAction<Exception> _onExceptionHandler;
    private ISerializer _serializer;
    private bool _disposed;

    public string FrontendConnectionString { get; }

    public string BackendConnectionString { get; }

    public PubSubMessageRouter(
        string backendConnectionString,
        string frontendConnectionString,
        string controlInConnectionName = "controlIn",
        string controlOutConnectionName = "controlOut")
    {
        BackendConnectionString = backendConnectionString;
        FrontendConnectionString = frontendConnectionString;
        _serializer = new Serialize.Json.JsonObjectSerializer();
        _xSub = new XSubscriberSocket($"@{FrontendConnectionString}");
        _xPub = new XPublisherSocket($"@{BackendConnectionString}");
        _xSub.Options.ReceiveHighWatermark = 10000;
        _xSub.Options.ReceiveBuffer = 44000;
        _xSub.Options.SendBuffer = 44000;
        _xSub.Options.Linger = TimeSpan.FromSeconds(5);
        _xPub.Options.ReceiveHighWatermark = 10000;
        _xPub.Options.ReceiveBuffer = 44000;
        _xPub.Options.SendBuffer = 44000;
        _xPub.Options.Linger = TimeSpan.FromSeconds(5);
        _controlInPush.Bind($"inproc://{controlInConnectionName}");
        _controlInPull.Connect($"inproc://{controlInConnectionName}");
        _controlOutPush.Bind($"inproc://{controlOutConnectionName}");
        _controlOutPull.Connect($"inproc://{controlOutConnectionName}");
        _poller = new NetMQPoller
        {
            _xSub,
            _xPub
        };
        // proxy messages between frontend / backend
        _proxy = new Proxy(
            _xSub,
            _xPub,
            poller: _poller,
            controlIn: _controlInPush,
            controlOut: _controlOutPush
        );
    }

    public PubSubMessageRouter(
        string backendConnectionString,
        string frontendConnectionString,
        Action<MessageBase> handlerReceived,
        Action<MessageBase> handlerSent,
        Action<Exception> onException=null) : this(backendConnectionString, frontendConnectionString)
    {
        _subscriberReceived = new WeakAction<MessageBase>(handlerReceived);
        _subscriberSent = new WeakAction<MessageBase>(handlerSent);

        if (onException != null)
        {
            _onExceptionHandler = new WeakAction<Exception>(onException);
        }
    }


    private async void MessagePumpReceived(CancellationToken ct)
    {
        while (true)
        {
            try
            {
                if (_cancellationToken.IsCancellationRequested)
                    break;

                var mqMessage = new NetMQMessage();

                if (_controlInPull.TryReceiveMultipartMessage(ref mqMessage))
                {
                    if (mqMessage.FrameCount == 1) //subscribe notification
                    {
                        continue;
                    }

                    var message = (MessageBase) _serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase));

                    if (_subscriberReceived?.Target != null)
                    {
                        _subscriberReceived.Execute(message);
                    }
                }
                else
                {
                    await Task.Delay(TimeSpan.FromMilliseconds(100), ct);
                }
            }
            catch (Exception ex)
            {
                OnExceptionHandler(ex);
            }
        }
    }

    private async void MessagePumpSent(CancellationToken ct)
    {
        while (true)
        {
            try
            {
                if (_cancellationToken.IsCancellationRequested)
                    break;

                var mqMessage = new NetMQMessage();

                if (_controlOutPull.TryReceiveMultipartMessage(ref mqMessage))
                {
                    if (mqMessage.FrameCount == 1) //subscribe notification
                    {
                        continue;
                    }


                    var message = (MessageBase) _serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase));

                    if (_subscriberSent?.Target != null)
                    {
                        _subscriberSent.Execute(message);
                    }
                }
                else
                {
                    await Task.Delay(TimeSpan.FromMilliseconds(100), ct);
                }
            }
            catch (Exception ex)
            {
                OnExceptionHandler(ex);
            }
        }
    }

    private void OnExceptionHandler(Exception obj)
    {
        if(_onExceptionHandler != null && _onExceptionHandler.IsAlive)
        {
            _onExceptionHandler.Execute(obj);
        }
    }

    public void StartRouting()
    {
        _cancellationTokenSource = new CancellationTokenSource();
        _cancellationToken = _cancellationTokenSource.Token;

        _proxy.Start();
        _poller.RunAsync();

        _workerReceived = Task.Run(
            () => MessagePumpReceived(_cancellationToken),
            _cancellationToken);

        _workerSent = Task.Run(
            () => MessagePumpSent(_cancellationToken),
            _cancellationToken);
    }

    public async Task StopRouting()
    {
        _proxy.Stop();
        _poller.Stop();

        _cancellationTokenSource?.Cancel();
        await Task.WhenAny(Task.WhenAll(_workerSent, _workerReceived),
            Task.Delay(5000)); //wait when all thread finished, but max 5 seconds
    }

    public async void Dispose(bool disposing)
    {
        if (_disposed)
            return;

        if (disposing)
        {
            await StopRouting();
            _poller?.Dispose();
            _poller = null;
            _proxy = null;
            _xSub?.Dispose();
            _xSub = null;
            _xPub?.Dispose();
            _xPub = null;
            _controlInPush?.Dispose();
            _controlInPush = null;
            _controlInPull?.Dispose();
            _controlInPull = null;
            _controlOutPush?.Dispose();
            _controlOutPush = null;
            _controlOutPull?.Dispose();
            _controlOutPull = null;
            _workerSent = null;
            _workerReceived = null;
            _cancellationTokenSource?.Dispose();
            _cancellationTokenSource = null;
            _subscriberReceived.MarkForDeletion();
            _subscriberReceived = null;
            _subscriberSent.MarkForDeletion();
            _subscriberSent = null;
            _onExceptionHandler.MarkForDeletion();
            _onExceptionHandler = null;
            _serializer = null;
        }

        _disposed = true;
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
        GC.Collect();
    }
}

I have ansyc message handlers which are fired by service api. Problem usually happens when i am sending 2 long messages one following other:

public async Task Handle(MessageBase message, IApiContext context)
{
using (var logger = new MessageLogger(typeof(RequestSuiteMetaDataHandler).FullName, context.ApplicationSettings.FrontendConnection, context.LogTopic))
using (var adapterManager = new AdaptersManager.AdaptersManager(new WindowsRegistrySettingsProvider()))
{

            var response = new RequestSuiteMetaDataResponse();

            try
            {
                var request = new JsonObjectSerializer().Deserialize<RequestSuiteMetaData>(message.Payload);
               //some code skipped
               
                response.Succeed = true;
                response.SuiteName = queryResponse.SuiteName;
                response.TestPhases = queryResponse.TestPhases;// long content around 8 kB
            catch (Exception exp)
            {
                await logger.LogMessage(LogLevel.Error, exp.GetAllMessages());
                response.Error = exp.GetAllMessages();
                response.Succeed = false;
            }

            var payloadString = new JsonObjectSerializer().SerializeToString(response);

            var messageToSend = new MessageBase
            {
                CorrelationId = message.CorrelationId,
                PayloadType = typeof(IRequestSuiteMetaDataResponse).FullName,
                SentTime = DateTime.Now,
                Payload = payloadString
            };

            // problem source:
            await logger.LogMessage(LogLevel.Debug,
                $"Sending message: {Base64Decode(messageToSend.Payload)} of type: {messageToSend.PayloadType}");
            Thread.Sleep(50);
            await context.ResponsePublisher.SendMessage(context.ResponseTopic, messageToSend);  
        }
    }

i have central async logger, sending log overzero mq.
public class MessageLogger: ILogger
{
const char TestBackSlash = '\';
const char TestSlash = '/';
const char TestDblQuote = '"';

    private readonly string _source;
    private readonly string _logTopic;
    private ISerializer _serializer;
    private IPublisher _publisher;
    private bool _disposed;

    public CdtMessageLogger(string source, 
        string frontendConnectionString, 
        string logTopic = TopicsList.AllLogs, 
        ISerializer serializer = null, 
        SingleThreadTaskScheduler scheduler = null)
    {
        _source = source;
        _logTopic = logTopic;
        _serializer = serializer ?? new Serialize.Json.JsonObjectSerializer();
        _publisher = new Publisher($">{frontendConnectionString}", scheduler);
    }

    public Task LogMessage(LogLevel level, string message)
    {
        var messageToSend = new MessageBase
        {
            CorrelationId = Guid.NewGuid(),
            PayloadType = typeof(ILogMessage).FullName,
            SentTime = DateTime.Now,
            Payload = _serializer.SerializeToString(new LogMessage
            {
                Level = level,
                Message = RemoveSpecialChars(message),
                Source = _source
            })
        };

        return _publisher.SendMessage(_logTopic, messageToSend);
    }

    private string RemoveSpecialChars(string input)
    {
        
        var output = new StringBuilder(input.Length);
        foreach (var c in input)
        {
            switch (c)
            {
                case TestSlash:
                    output.AppendFormat("{0}{1}", TestBackSlash, TestSlash);
                    break;

                case TestBackSlash:
                    output.AppendFormat("{0}{0}", TestBackSlash);
                    break;

                case TestDblQuote:
                    output.AppendFormat("{0}{1}", TestBackSlash, TestDblQuote);
                    break;

                case '\b':
                    output.Append("\\b");
                    break;

                case '\f':
                    output.Append("\\f");
                    break;

                case '\n':
                    output.Append("\\n");
                    break;

                case '\r':
                    output.Append("\\r");
                    break;

                case '\t':
                    output.Append("\\t");
                    break;

                default:
                    output.Append(c);
                    break;
            }
        }

        return output.ToString();
    }


    public void Dispose(bool disposing)
    {
        if (_disposed)
            return;

        if (disposing)
        {
            _publisher?.Dispose();
            _publisher = null;
            _serializer = null;
        }

        _disposed = true;
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
        GC.Collect();
    }
}

public class Publisher: IPublisher
{
private PublisherSocket _pub;
private ISerializer _serializer;
private SingleThreadTaskScheduler _taskScheduler;
private bool _disposed;

    public Publisher(string connectionString , SingleThreadTaskScheduler scheduler=null)
    {
        _pub = new PublisherSocket(connectionString);
        _pub.Options.SendHighWatermark = 1000;
        _pub.Options.ReceiveBuffer = 44000;
        _pub.Options.SendBuffer = 44000;
        _pub.Options.ReconnectInterval = TimeSpan.FromMilliseconds(50);
        _pub.Options.Linger = TimeSpan.FromSeconds(5);
        _serializer = new Serialize.Json.JsonObjectSerializer();
        _taskScheduler = scheduler ?? new SingleThreadTaskScheduler();
    }

    public void Dispose(bool disposing)
    {
        if (_disposed)
            return;

        if (disposing)
        {
            _pub?.Close();
            _pub?.Dispose();
            _pub = null;
            _serializer = null;
            _taskScheduler.StopSchedulerThread();
            _taskScheduler.Dispose();
            _taskScheduler = null;
        }

        _disposed = true;
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
        GC.Collect();
    }

    public Task SendMessage(string topicName, MessageBase message)
    {
        return Task.Factory.StartNew(() =>
        {
            var data = _serializer.Serialize(message);
            var messToSend = new NetMQMessage();
            messToSend.Append(topicName);
            messToSend.Append(data);
            _pub.SendMultipartMessage(messToSend);
        }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler);
    }
}

Message rate is not so high <20 messages/s, usually a lot of short messages and sometimes a long ones. It is looking like message buffer is allocated in correct size, but not all characters are received, why?

any ideas?

every part of the system, router, central logger, handlers are living in own windows service.

Any chance that you are re-using the buffer when sending the message? That might be the problem.

…

On Fri, Nov 8, 2019, 09:35 m786-tech ***@***.***> wrote: every part of the system, router, central logger, handlers are living in own windows service. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAUW75UVA3E3XAJEOB7UANDQSUJCTA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDO73JI#issuecomment-551419301>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAUW75VASIHVMHYRJ466GETQSUJCTANCNFSM4JKTUDTA> .

Speaking from experience, I've run into that problem before. To verify this is not the issue, wrap the message in a checksum, and generate/check this checksum in debug mode. If the checksum is anything other than expected, it will pin down the source of the issue.

…

On Fri, 8 Nov 2019, 17:06 Doron Somech, ***@***.***> wrote: Any chance that you are re-using the buffer when sending the message? That might be the problem. On Fri, Nov 8, 2019, 09:35 m786-tech ***@***.***> wrote: > every part of the system, router, central logger, handlers are living in > own windows service. > > — > You are receiving this because you are subscribed to this thread. > Reply to this email directly, view it on GitHub > < #14?email_source=notifications&email_token=AAUW75UVA3E3XAJEOB7UANDQSUJCTA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDO73JI#issuecomment-551419301 >, > or unsubscribe > < https://github.com/notifications/unsubscribe-auth/AAUW75VASIHVMHYRJ466GETQSUJCTANCNFSM4JKTUDTA > > . > — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAJ3FJEVNEAL22WG7V63WHTQSWL6TA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDSXWHI#issuecomment-551910173>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJ3FJG53N646MWEGNNQPYTQSWL6TANCNFSM4JKTUDTA> .

Any chance that you are re-using the buffer when sending the message? That might be the problem.
…
On Fri, Nov 8, 2019, 09:35 m786-tech @.***> wrote: every part of the system, router, central logger, handlers are living in own windows service. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAUW75UVA3E3XAJEOB7UANDQSUJCTA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDO73JI#issuecomment-551419301>, or unsubscribe https://github.com/notifications/unsubscribe-auth/AAUW75VASIHVMHYRJ466GETQSUJCTANCNFSM4JKTUDTA .

Hi, maybe it was not clear, but i am talking about buffer of NetMQMessage object.
I am waiting in TryReceiveMultipartMessage method until it return a NetMQMessage object.
Then using a last frame to deserialize a message. A buffer of that last frame should contain base64 UTF-8 bytes, but sometimes it contains some zeros on the end.

var mqMessage = new NetMQMessage();
// some code skipped
 var messReceived = (MessageBase) serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase));

I am always creating a new NetMQMessage object. I did removed Timeout in TryReceiveMultipartMessage method for debugging. I just waiting in that part of code until something received, without any break.
Busically i expecting that when it returns from TryReceiveMultipartMessage method, then reading is finished and complete right?But sometimes it is not true. Buffer in the frame is allocated but some bytes on the end are missing

public class RequestResponseClient: IRequestResponseClient
    {
        public Task<MessageBase> SendRequest(
            string frontendConnectionString, 
            string backendConnectionString, 
            MessageBase message, 
            string requestTopic, 
            string responseTopic, 
            TimeSpan timeout, 
            CancellationToken ct = default)
        {
            return SendRequest(frontendConnectionString, backendConnectionString, message, requestTopic, responseTopic,
                timeout, null, ct);

        }

        public Task<MessageBase> SendRequest(
            string frontendConnectionString, 
            string backendConnectionString, 
            MessageBase message, 
            string requestTopic, 
            string responseTopic, 
            TimeSpan timeout, 
            string expectedResponseType,
            CancellationToken ct = default)
        {
            return Task.Factory.StartNew(() =>
            {
                using (var publisher = new Publisher($">{frontendConnectionString}"))
                using (var subscriber = new SubscriberSocket($">{backendConnectionString}"))
                {
                    subscriber.Options.ReceiveHighWatermark = 1000;
                    subscriber.Options.ReceiveBuffer = 44000;
                    subscriber.Options.SendBuffer = 44000;
                    subscriber.Options.Linger = TimeSpan.FromSeconds(5);
                    Thread.Sleep(200);// give publisher change to learn subscriptions
                    subscriber.Subscribe(responseTopic);
                    var serializer = new Serialize.Json.JsonObjectSerializer();
                    publisher.SendMessage(requestTopic, message);
                    var sw = new Stopwatch();
                    sw.Start();

                    while (true)
                    {


                        if (ct.IsCancellationRequested)
                        {
                            subscriber.Unsubscribe(responseTopic);
                            ct.ThrowIfCancellationRequested();
                        }

                        var mqMessage = new NetMQMessage();

                        if (subscriber.TryReceiveMultipartMessage(ref mqMessage))
                        {
                            if (mqMessage.FrameCount == 1) //subscribe notification
                                continue;

                            var messReceived =
                                (MessageBase) serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase));

                            if (messReceived.CorrelationId == message.CorrelationId)
                            {
                                if (string.IsNullOrEmpty(expectedResponseType)) // if no filter, return message
                                    return messReceived;

                                if (messReceived.PayloadType == expectedResponseType)
                                    return messReceived;
                            }

                        }
                        else
                        {
                            Thread.Sleep(10);
                        }

                        if (sw.ElapsedMilliseconds > timeout.TotalMilliseconds)
                        {
                            subscriber.Unsubscribe(responseTopic);
                            throw new TimeoutException();
                        }


                    }

                }
            }, ct);

        }
    }

Understood, but it might be (probably) lost on the send side. So this is why I'm asking if by any chance you are re using the buffer after sending.

…

On Fri, Nov 8, 2019, 20:20 m786-tech ***@***.***> wrote: Any chance that you are re-using the buffer when sending the message? That might be the problem. … <#m_-3278168522527536637_> On Fri, Nov 8, 2019, 09:35 m786-tech *@*.***> wrote: every part of the system, router, central logger, handlers are living in own windows service. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#14 <#14>?email_source=notifications&email_token=AAUW75UVA3E3XAJEOB7UANDQSUJCTA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDO73JI#issuecomment-551419301>, or unsubscribe https://github.com/notifications/unsubscribe-auth/AAUW75VASIHVMHYRJ466GETQSUJCTANCNFSM4JKTUDTA . Hi, maybe it was not clear, but i am talking about buffer of NetMQMessage object. I am waiting in TryReceiveMultipartMessage method until it return a NetMQMessage object. Then using a last frame to deserialize a message. A buffer of that last frame should contain base64 UTF-8 bytes, but sometimes it contains some zeros on the end. var mqMessage = new NetMQMessage(); // some code skipped var messReceived = (MessageBase) serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase)); I am always creating a new NetMQMessage object. I did removed Timeout in TryReceiveMultipartMessage method for debugging. I just waiting in that part of code until something received, without any break. Busically i expecting that when it returns from TryReceiveMultipartMessage method, then reading is finished and complete right?But sometimes it is not true. Buffer in the frame is allocated but some bytes on the end are missing public class RequestResponseClient: IRequestResponseClient { public Task<MessageBase> SendRequest( string frontendConnectionString, string backendConnectionString, MessageBase message, string requestTopic, string responseTopic, TimeSpan timeout, CancellationToken ct = default) { return SendRequest(frontendConnectionString, backendConnectionString, message, requestTopic, responseTopic, timeout, null, ct); } public Task<MessageBase> SendRequest( string frontendConnectionString, string backendConnectionString, MessageBase message, string requestTopic, string responseTopic, TimeSpan timeout, string expectedResponseType, CancellationToken ct = default) { return Task.Factory.StartNew(() => { using (var publisher = new Publisher($">{frontendConnectionString}")) using (var subscriber = new SubscriberSocket($">{backendConnectionString}")) { subscriber.Options.ReceiveHighWatermark = 1000; subscriber.Options.ReceiveBuffer = 44000; subscriber.Options.SendBuffer = 44000; subscriber.Options.Linger = TimeSpan.FromSeconds(5); Thread.Sleep(200);// give publisher change to learn subscriptions subscriber.Subscribe(responseTopic); var serializer = new Serialize.Json.JsonObjectSerializer(); publisher.SendMessage(requestTopic, message); var sw = new Stopwatch(); sw.Start(); while (true) { if (ct.IsCancellationRequested) { subscriber.Unsubscribe(responseTopic); ct.ThrowIfCancellationRequested(); } var mqMessage = new NetMQMessage(); if (subscriber.TryReceiveMultipartMessage(ref mqMessage)) { if (mqMessage.FrameCount == 1) //subscribe notification continue; var messReceived = (MessageBase) serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase)); if (messReceived.CorrelationId == message.CorrelationId) { if (string.IsNullOrEmpty(expectedResponseType)) // if no filter, return message return messReceived; if (messReceived.PayloadType == expectedResponseType) return messReceived; } } else { Thread.Sleep(10); } if (sw.ElapsedMilliseconds > timeout.TotalMilliseconds) { subscriber.Unsubscribe(responseTopic); throw new TimeoutException(); } } } }, ct); } } — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAUW75UV56SUOPUJUK7PQI3QSWUXHA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDS6C6A#issuecomment-551936376>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAUW75WLBE6XMFQSL3RATK3QSWUXHANCNFSM4JKTUDTA> .

Hi,
I am always creating a new NetMQMessage object when sending a message.
What i found that it happens ususally when i am sending message twice, once to log topic and then to response topic. Logger class is sending it to logger topic and publisher to response topic.
They are both created on demand inside using block. You can see example above in RequestResponseClient.
Logs from Router already contains a corrupted base64 string. So if router is just distributing message it is distributed corrupted and RequestResponseClient receive it corrupted as well.

So as you mentioned problem could be when sending or maybe also router receive it incomplete?
Below is a publisher code. Task is always executed in the the SingleThreadTaskScheduler, which was created on same thread as publisher socket.

I see the only option is to use wireshark to see content.

// publisher class
public Task SendMessage(string topicName, MessageBase message)
    {
        return Task.Factory.StartNew(() =>
        {
            var data = _serializer.Serialize(message);
            var messToSend = new NetMQMessage();
            messToSend.Append(topicName);
            messToSend.Append(data);
            _pub.SendMultipartMessage(messToSend);
        }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler);
    }

//service method handler class
public async Task Handle(MessageBase message, IApiContext context)
{
using (var logger = new MessageLogger(typeof(RequestSuiteMetaDataHandler).FullName, context.ApplicationSettings.FrontendConnection, context.LogTopic))
using (var adapterManager = new AdaptersManager.AdaptersManager(new WindowsRegistrySettingsProvider()))
{

            var response = new RequestSuiteMetaDataResponse();

            try
            {
                var request = new JsonObjectSerializer().Deserialize<RequestSuiteMetaData>(message.Payload);
               //some code skipped
               
                response.Succeed = true;
                response.SuiteName = queryResponse.SuiteName;
                response.TestPhases = queryResponse.TestPhases;// long content around 8 kB
            catch (Exception exp)
            {
                await logger.LogMessage(LogLevel.Error, exp.GetAllMessages());
                response.Error = exp.GetAllMessages();
                response.Succeed = false;
            }

            var payloadString = new JsonObjectSerializer().SerializeToString(response);

            var messageToSend = new MessageBase
            {
                CorrelationId = message.CorrelationId,
                PayloadType = typeof(IRequestSuiteMetaDataResponse).FullName,
                SentTime = DateTime.Now,
                Payload = payloadString
            };

            // problem source:
            await logger.LogMessage(LogLevel.Debug,
                $"Sending message: {Base64Decode(messageToSend.Payload)} of type: {messageToSend.PayloadType}");
            Thread.Sleep(50);
            await context.ResponsePublisher.SendMessage(context.ResponseTopic, messageToSend);  
        }
    }

Problem disappeared once i removed this part:

 await logger.LogMessage(LogLevel.Debug,
                $"Sending message: {Base64Decode(messageToSend.Payload)} of type: {messageToSend.PayloadType}");

From the snippet you just send it seems you are using netmq from multiple threads, which is not supported. You can only use a socket from one thread, that includes sending, receiving or polling. All must be from some thread.

…

On Sat, Nov 9, 2019, 10:57 m786-tech ***@***.***> wrote: Hi, I am always creating a new NetMQMessage object when sending a message. What i found that it happens ususally when i am sending message twice, once to log topic and then to response topic. Logger class is sending it to logger topic and publisher to response topic. They are both created on demand inside using block. You can see example above in RequestResponseClient. Logs from Router already contains a corrupted base64 string. So if router is just distributing message it is distributed corrupted and RequestResponseClient receive it corrupted as well. So as you mentioned problem could be when sending or maybe also router receive it incomplete? Below is a publisher code. Task is always executed in the the SingleThreadTaskScheduler, which was created on same thread as publisher socket. Can i somehow enable debugging on socket level in NetMQ library? I will try to use wireshark to see content. public Task SendMessage(string topicName, MessageBase message) { return Task.Factory.StartNew(() => { var data = _serializer.Serialize(message); var messToSend = new NetMQMessage(); messToSend.Append(topicName); messToSend.Append(data); _pub.SendMultipartMessage(messToSend); }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler); } — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAUW75V5MBQ5VLPTSMVQRTLQSZ3QNA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDUBITQ#issuecomment-552080462>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAUW75UPXE2H2SDP3YKZ2KLQSZ3QNANCNFSM4JKTUDTA> .

Try to drop the Task.Factory.StartNew, just send from that thread. Send takes zero time if socket is not full.

…

On Sat, Nov 9, 2019, 15:52 Doron Somech ***@***.***> wrote: From the snippet you just send it seems you are using netmq from multiple threads, which is not supported. You can only use a socket from one thread, that includes sending, receiving or polling. All must be from some thread. On Sat, Nov 9, 2019, 10:57 m786-tech ***@***.***> wrote: > Hi, > I am always creating a new NetMQMessage object when sending a message. > What i found that it happens ususally when i am sending message twice, > once to log topic and then to response topic. Logger class is sending it to > logger topic and publisher to response topic. > They are both created on demand inside using block. You can see example > above in RequestResponseClient. > Logs from Router already contains a corrupted base64 string. So if router > is just distributing message it is distributed corrupted and > RequestResponseClient receive it corrupted as well. > > So as you mentioned problem could be when sending or maybe also router > receive it incomplete? > Below is a publisher code. Task is always executed in the the > SingleThreadTaskScheduler, which was created on same thread as publisher > socket. > > Can i somehow enable debugging on socket level in NetMQ library? > I will try to use wireshark to see content. > > public Task SendMessage(string topicName, MessageBase message) > { > return Task.Factory.StartNew(() => > { > var data = _serializer.Serialize(message); > var messToSend = new NetMQMessage(); > messToSend.Append(topicName); > messToSend.Append(data); > _pub.SendMultipartMessage(messToSend); > }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler); > } > > — > You are receiving this because you commented. > Reply to this email directly, view it on GitHub > <#14?email_source=notifications&email_token=AAUW75V5MBQ5VLPTSMVQRTLQSZ3QNA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDUBITQ#issuecomment-552080462>, > or unsubscribe > <https://github.com/notifications/unsubscribe-auth/AAUW75UPXE2H2SDP3YKZ2KLQSZ3QNANCNFSM4JKTUDTA> > . >

My best guess on fixing this is to have all producers adding copies of the messages to a thread safe queue, then have a dedicated queue which sends to NetMQ. If the message is a reference type, and it is mutable (not immutable), you may have to create a copy each message before placing it into the queue. See answer from Contango here: https://stackoverflow.com/questions/129389/how-do-you-do-a-deep-copy-of-an-object-in-net-c-specifically

…

On Sat, 9 Nov 2019, 13:53 Doron Somech, ***@***.***> wrote: From the snippet you just send it seems you are using netmq from multiple threads, which is not supported. You can only use a socket from one thread, that includes sending, receiving or polling. All must be from some thread. On Sat, Nov 9, 2019, 10:57 m786-tech ***@***.***> wrote: > Hi, > I am always creating a new NetMQMessage object when sending a message. > What i found that it happens ususally when i am sending message twice, > once to log topic and then to response topic. Logger class is sending it to > logger topic and publisher to response topic. > They are both created on demand inside using block. You can see example > above in RequestResponseClient. > Logs from Router already contains a corrupted base64 string. So if router > is just distributing message it is distributed corrupted and > RequestResponseClient receive it corrupted as well. > > So as you mentioned problem could be when sending or maybe also router > receive it incomplete? > Below is a publisher code. Task is always executed in the the > SingleThreadTaskScheduler, which was created on same thread as publisher > socket. > > Can i somehow enable debugging on socket level in NetMQ library? > I will try to use wireshark to see content. > > public Task SendMessage(string topicName, MessageBase message) > { > return Task.Factory.StartNew(() => > { > var data = _serializer.Serialize(message); > var messToSend = new NetMQMessage(); > messToSend.Append(topicName); > messToSend.Append(data); > _pub.SendMultipartMessage(messToSend); > }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler); > } > > — > You are receiving this because you commented. > Reply to this email directly, view it on GitHub > < #14?email_source=notifications&email_token=AAUW75V5MBQ5VLPTSMVQRTLQSZ3QNA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDUBITQ#issuecomment-552080462 >, > or unsubscribe > < https://github.com/notifications/unsubscribe-auth/AAUW75UPXE2H2SDP3YKZ2KLQSZ3QNANCNFSM4JKTUDTA > > . > — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAJ3FJH5YOTQ25Y6GTBH76LQS26EBA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDUGUHY#issuecomment-552102431>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJ3FJGUX6U3ZAQ4BPRHEPDQS26EBANCNFSM4JKTUDTA> .

Hi guys,
Thanks for answears.
As far i know, i am doing all operations with single socket on same thread.I have SingleThreadTaskScheduler class. SingleThreadTaskScheduler class have one thread and execute every task on that thread.
I never got a exception like access from another thread. But i will recheck it.

SingleThreadTaskScheduler instance is passed as parameter into Task.Factory.StartNew.
So Task created by this method is scheduled by my own scheduler, which des not use ThreadPool, but always same thread.

I will recheck the code and see.

HI guys, it is looking that TaskScheduler.FromCurrentSynchronizationContext() is not always working as i expected. I used it in some components inside task, to create instances and run code on same thread.
i have changed Subscriber and publisher classes to be sure that same thread scheduler used every time.

i have changed constructors to be private and introduced factory methods.
Task scheduler created in that factory method is used also to create all sockets instances and later on for all calls. Currently it looks no more problems, but i will test it more. Funny thing is that i never got a exception about something illegal accessing or so on.

    public class Publisher: IPublisher
    {
        private PublisherSocket _pub;
        private ISerializer _serializer;
        private TaskScheduler _taskScheduler;
        private bool _disposed;

        public static async Task<Publisher> CreateAsync(string connectionString, TaskScheduler scheduler = null)
        {
            var publisher = new Publisher();
            await publisher.Initialize(connectionString, scheduler);
            return publisher;
        }

        private Publisher()
        {
            
        }

        private async Task Initialize(string connectionString, TaskScheduler scheduler = null)
        {
            _taskScheduler = scheduler ?? new ThreadTaskScheduler();

            _pub = await Task.Factory.StartNew(() => 
                new PublisherSocket(connectionString), CancellationToken.None, TaskCreationOptions.None, _taskScheduler);
            _pub.Options.SendHighWatermark = 1000;
            _pub.Options.ReceiveBuffer = 44000;
            _pub.Options.SendBuffer = 44000;
            _pub.Options.ReconnectInterval = TimeSpan.FromMilliseconds(50);
            _pub.Options.Linger = TimeSpan.FromSeconds(5);
            _serializer = new Serialize.Json.JsonObjectSerializer();
            
        }

        public void Dispose(bool disposing)
        {
            if (_disposed)
                return;

            if (disposing)
            {
                _pub?.Close();
                _pub?.Dispose();
                _pub = null;
                _serializer = null;
                _taskScheduler = null;
            }

            _disposed = true;
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
            GC.Collect();
        }

        public Task SendMessage(string topicName, MessageBase message)
        {
            return Task.Factory.StartNew(() =>
            {
                var data = _serializer.Serialize(message);
                var messToSend = new NetMQMessage();
                messToSend.Append(topicName);
                messToSend.Append(data);
                _pub.SendMultipartMessage(messToSend);
            }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler);
        }
    }

public class Subscriber : ISubscriber
    {
        private string _topicName;
        private SubscriberSocket _sub;
        private ISerializer _serializer;
        private Task _worker;
        private CancellationTokenSource _cancellationTokenSource;
        private CancellationToken _cancellationToken;
        private WeakAction<MessageBase> _subscriber;
        private WeakAction<Exception> _onException;
        private TaskScheduler _taskScheduler;
        private bool _disposed;

        public static async Task<Subscriber> CreateAsync(
            string connectionString,
            string topicName,
            Action<MessageBase> handler,
            object handlerTarget,
            Action<Exception> onException = null,
            object onExceptionTarget = null,
            TaskScheduler scheduler = null)
        {
            var sub = new Subscriber();
            await sub.Initialize(connectionString, topicName, handler, handlerTarget, onException, onExceptionTarget);
            return sub;
        }

        private Subscriber()
        { }

        private async  Task Initialize(
            string connectionString,
            string topicName,
            Action<MessageBase> handler,
            object handlerTarget,
            Action<Exception> onException = null,
            object onExceptionTarget = null,
            TaskScheduler scheduler = null)
        {
            _taskScheduler = scheduler ?? new ThreadTaskScheduler();

            _sub = await Task.Factory.StartNew(() =>
                new SubscriberSocket(connectionString), CancellationToken.None, TaskCreationOptions.None, _taskScheduler);

            _topicName = topicName;
            _sub.Options.ReceiveHighWatermark = 1000;
            _sub.Options.ReceiveBuffer = 44000;
            _sub.Options.SendBuffer = 44000;
            _sub.Options.Linger = TimeSpan.FromSeconds(5);

            if (onException != null)
            {
                _onException = new WeakAction<Exception>(onExceptionTarget, onException);
            }

            _serializer = new Serialize.Json.JsonObjectSerializer();
            _subscriber = new WeakAction<MessageBase>(handlerTarget, handler);
            _sub.Subscribe(topicName);
            _cancellationTokenSource = new CancellationTokenSource();
            _cancellationToken = _cancellationTokenSource.Token;
            _worker = Task.Factory.StartNew(
                () => MessagePump(_cancellationToken),
                _cancellationToken, TaskCreationOptions.None, _taskScheduler);
        }

        public void Dispose(bool disposing)
        {
            if (_disposed)
                return;

            if (disposing)
            {
                _cancellationTokenSource?.Cancel();
                Task.WaitAll(_worker);
                _worker = null;
                _cancellationTokenSource?.Dispose();
                _cancellationTokenSource = null;
                _serializer = null;
                _sub.Unsubscribe(_topicName);
                _subscriber.MarkForDeletion();
                _onException?.MarkForDeletion();
                _onException = null;
                _sub.Close();
                _sub.Dispose();
                _sub = null;
                _subscriber = null;
            }

            _disposed = true;
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
            GC.Collect();
        }

        private void MessageRouter_OnException(Exception obj)
        {
            if(_onException!= null && _onException.IsAlive)
                _onException.Execute(obj);
        }

        private async void MessagePump(CancellationToken ct)
        {
            while (true)
            {
                try
                {
                    if (_cancellationToken.IsCancellationRequested)
                        break;

                    var mqMessage = new NetMQMessage(2);

                    if (_sub.TryReceiveMultipartMessage(ref mqMessage, 2))
                    {
                        if (mqMessage.FrameCount == 1) //subscribe notification
                        {
                            continue;

                        }

                        var message = (MessageBase) _serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase));


                        if (_subscriber.Target != null)
                        {
                            _subscriber.Execute(message);
                        }
                    }
                    else
                    {
                        await Task.Delay(TimeSpan.FromMilliseconds(100), ct);
                    }
                }
                catch (Exception e)
                {
                    MessageRouter_OnException(e);
                }
            }
        }
    }
    ```

Glad you have a solution. What version of .NET are you running?

…

On Mon, 11 Nov 2019, 13:34 m786-tech, ***@***.***> wrote: HI guys, it is looking that TaskScheduler.FromCurrentSynchronizationContext() is not always working as i expected. I used it in some components inside task, to create instances and run code on same thread. i have changed Subscriber and publisher classes to be sure that same thread scheduler used every time. i have changed constructors to be private and introduced factory methods. Task scheduler created in that factory method is used also to create all sockets instances and later on for all calls. Currently it looks no more problems, but i will test it more. Funny thing is that i never got a exception about something illegal accessing or so on. public class Publisher: IPublisher { private PublisherSocket _pub; private ISerializer _serializer; private TaskScheduler _taskScheduler; private bool _disposed; public static async Task<Publisher> CreateAsync(string connectionString, TaskScheduler scheduler = null) { var publisher = new Publisher(); await publisher.Initialize(connectionString, scheduler); return publisher; } private Publisher() { } private async Task Initialize(string connectionString, TaskScheduler scheduler = null) { _taskScheduler = scheduler ?? new ThreadTaskScheduler(); _pub = await Task.Factory.StartNew(() => new PublisherSocket(connectionString), CancellationToken.None, TaskCreationOptions.None, _taskScheduler); _pub.Options.SendHighWatermark = 1000; _pub.Options.ReceiveBuffer = 44000; _pub.Options.SendBuffer = 44000; _pub.Options.ReconnectInterval = TimeSpan.FromMilliseconds(50); _pub.Options.Linger = TimeSpan.FromSeconds(5); _serializer = new Serialize.Json.JsonObjectSerializer(); } public void Dispose(bool disposing) { if (_disposed) return; if (disposing) { _pub?.Close(); _pub?.Dispose(); _pub = null; _serializer = null; _taskScheduler = null; } _disposed = true; } public void Dispose() { Dispose(true); GC.SuppressFinalize(this); GC.Collect(); } public Task SendMessage(string topicName, MessageBase message) { return Task.Factory.StartNew(() => { var data = _serializer.Serialize(message); var messToSend = new NetMQMessage(); messToSend.Append(topicName); messToSend.Append(data); _pub.SendMultipartMessage(messToSend); }, CancellationToken.None, TaskCreationOptions.None, _taskScheduler); } } public class Subscriber : ISubscriber { private string _topicName; private SubscriberSocket _sub; private ISerializer _serializer; private Task _worker; private CancellationTokenSource _cancellationTokenSource; private CancellationToken _cancellationToken; private WeakAction<MessageBase> _subscriber; private WeakAction<Exception> _onException; private TaskScheduler _taskScheduler; private bool _disposed; public static async Task<Subscriber> CreateAsync( string connectionString, string topicName, Action<MessageBase> handler, object handlerTarget, Action<Exception> onException = null, object onExceptionTarget = null, TaskScheduler scheduler = null) { var sub = new Subscriber(); await sub.Initialize(connectionString, topicName, handler, handlerTarget, onException, onExceptionTarget); return sub; } private Subscriber() { } private async Task Initialize( string connectionString, string topicName, Action<MessageBase> handler, object handlerTarget, Action<Exception> onException = null, object onExceptionTarget = null, TaskScheduler scheduler = null) { _taskScheduler = scheduler ?? new ThreadTaskScheduler(); _sub = await Task.Factory.StartNew(() => new SubscriberSocket(connectionString), CancellationToken.None, TaskCreationOptions.None, _taskScheduler); _topicName = topicName; _sub.Options.ReceiveHighWatermark = 1000; _sub.Options.ReceiveBuffer = 44000; _sub.Options.SendBuffer = 44000; _sub.Options.Linger = TimeSpan.FromSeconds(5); if (onException != null) { _onException = new WeakAction<Exception>(onExceptionTarget, onException); } _serializer = new Serialize.Json.JsonObjectSerializer(); _subscriber = new WeakAction<MessageBase>(handlerTarget, handler); _sub.Subscribe(topicName); _cancellationTokenSource = new CancellationTokenSource(); _cancellationToken = _cancellationTokenSource.Token; _worker = Task.Factory.StartNew( () => MessagePump(_cancellationToken), _cancellationToken, TaskCreationOptions.None, _taskScheduler); } public void Dispose(bool disposing) { if (_disposed) return; if (disposing) { _cancellationTokenSource?.Cancel(); Task.WaitAll(_worker); _worker = null; _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; _serializer = null; _sub.Unsubscribe(_topicName); _subscriber.MarkForDeletion(); _onException?.MarkForDeletion(); _onException = null; _sub.Close(); _sub.Dispose(); _sub = null; _subscriber = null; } _disposed = true; } public void Dispose() { Dispose(true); GC.SuppressFinalize(this); GC.Collect(); } private void MessageRouter_OnException(Exception obj) { if(_onException!= null && _onException.IsAlive) _onException.Execute(obj); } private async void MessagePump(CancellationToken ct) { while (true) { try { if (_cancellationToken.IsCancellationRequested) break; var mqMessage = new NetMQMessage(2); if (_sub.TryReceiveMultipartMessage(ref mqMessage, 2)) { if (mqMessage.FrameCount == 1) //subscribe notification { continue; } var message = (MessageBase) _serializer.Deserialize(mqMessage.Last.Buffer, typeof(MessageBase)); if (_subscriber.Target != null) { _subscriber.Execute(message); } } else { await Task.Delay(TimeSpan.FromMilliseconds(100), ct); } } catch (Exception e) { MessageRouter_OnException(e); } } } } ``` — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAJ3FJASLBQILM4CQCQGKTTQTFNKRA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDW225Y#issuecomment-552447351>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJ3FJFMIA6W4KLDPU6FZMLQTFNKRANCNFSM4JKTUDTA> .

.net core 2.2

Interesting. Wonder if the problem would disappear on .NET Core V3? This function call has been present since .NET 4. I wonder if there are any other reports of instability with this call?

…

On Mon, 11 Nov 2019, 14:05 m786-tech, ***@***.***> wrote: .net core 2.2 — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#14?email_source=notifications&email_token=AAJ3FJFNNDMST7LA5VFNPGDQTFQ7DA5CNFSM4JKTUDTKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEDW5QMY#issuecomment-552458291>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJ3FJDIT5OMZWERX6II3KLQTFQ7DANCNFSM4JKTUDTA> .

Not sure, not sure also if problem is exactly that call, or the way how it was used.
I did use net mq for parallel/multithread tasks.It is much easier to use some messaging system as to use kind of synchronization or locks or other primitives. I mean, situation where every component want to log something without waiting, is messaging working perfectly. Just sending message into publisher socket and going on with other taks.On other side subscribers/workers will collect all the logs and store them into file, which takes some time. Another use case i have is service api, which fire up new handler instance for every request, then i can handle a lot of requests without long waiting times. Every handler want to publish, or log something without waiting.This is easy if i create new short lifetime publisher which just publish message to worker topic and go away....

Navigation Menu

Repository navigation

message cutting #14

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

message cutting #14

Description

Activity

m786-tech commented on Nov 8, 2019

somdoron commented on Nov 8, 2019

sharpe5 commented on Nov 8, 2019

m786-tech commented on Nov 8, 2019

somdoron commented on Nov 8, 2019

m786-tech commented on Nov 9, 2019

somdoron commented on Nov 9, 2019

somdoron commented on Nov 9, 2019

sharpe5 commented on Nov 9, 2019

m786-tech commented on Nov 9, 2019

m786-tech commented on Nov 11, 2019

sharpe5 commented on Nov 11, 2019

m786-tech commented on Nov 11, 2019

sharpe5 commented on Nov 11, 2019

m786-tech commented on Nov 11, 2019

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions