<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

namespace Dbus
{
    public partial class Connection : IDisposable
    {
        public const string SystemBusAddress = "unix:path=/var/run/dbus/system_bus_socket";

        private readonly Stream stream;
        private readonly ConcurrentDictionary<uint, TaskCompletionSource<ReceivedMethodReturn>> expectedMessages;
        private readonly ConcurrentDictionary<string, Action<MessageHeader, byte[]>> signalHandlers;
        private readonly ConcurrentDictionary<string, Func<uint, MessageHeader, byte[], Task>> objectProxies;
        private readonly CancellationTokenSource receiveCts;

        private int serialCounter;

        private Connection(Stream stream)
        {
            this.stream = stream;

            expectedMessages = new ConcurrentDictionary<uint, TaskCompletionSource<ReceivedMethodReturn>>();
            signalHandlers = new ConcurrentDictionary<string, Action<MessageHeader, byte[]>>();
            objectProxies = new ConcurrentDictionary<string, Func<uint, MessageHeader, byte[], Task>>();
            receiveCts = new CancellationTokenSource();

#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
            // Ideally, there would be a DisposeAsync to properly await the receive task.
            // It's stopped properly, though
            Task.Run(receive);
#pragma warning restore CS4014
        }

        public async static Task<Connection> CreateAsync(string address)
        {
            var endPoint = EndPointFactory.Create(address);
            var socket = new Socket(AddressFamily.Unix, SocketType.Stream, 0);
            await socket.ConnectAsync(endPoint).ConfigureAwait(false);
            var stream = new LoggingStream(new NetworkStream(socket, ownsSocket: true));

            await authenticate(stream).ConfigureAwait(false);

            return new Connection(stream);
        }

        public IDisposable RegisterObjectProxy(
            ObjectPath path,
            string interfaceName,
            Func<uint, MessageHeader, byte[], Task> proxy
        )
        {
            var dictionaryEntry = path + "\0" + interfaceName;
            if (!objectProxies.TryAdd(dictionaryEntry, proxy))
                throw new InvalidOperationException("Attempted to register an object proxy twice");

            return new deregistration
            {
                Deregister = () =>
                {
                    Func<uint, MessageHeader, byte[], Task> _;
                    objectProxies.TryRemove(dictionaryEntry, out _);
                }
            };
        }

        public IDisposable RegisterSignalHandler(
            ObjectPath path,
            string interfaceName,<commit_msg>Disable logging the bytes traveling the wire<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

namespace Dbus
{
    public partial class Connection : IDisposable
    {
        public const string SystemBusAddress = "unix:path=/var/run/dbus/system_bus_socket";

        private readonly Stream stream;
        private readonly ConcurrentDictionary<uint, TaskCompletionSource<ReceivedMethodReturn>> expectedMessages;
        private readonly ConcurrentDictionary<string, Action<MessageHeader, byte[]>> signalHandlers;
        private readonly ConcurrentDictionary<string, Func<uint, MessageHeader, byte[], Task>> objectProxies;
        private readonly CancellationTokenSource receiveCts;

        private int serialCounter;

        private Connection(Stream stream)
        {
            this.stream = stream;

            expectedMessages = new ConcurrentDictionary<uint, TaskCompletionSource<ReceivedMethodReturn>>();
            signalHandlers = new ConcurrentDictionary<string, Action<MessageHeader, byte[]>>();
            objectProxies = new ConcurrentDictionary<string, Func<uint, MessageHeader, byte[], Task>>();
            receiveCts = new CancellationTokenSource();

#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
            // Ideally, there would be a DisposeAsync to properly await the receive task.
            // It's stopped properly, though
            Task.Run(receive);
#pragma warning restore CS4014
        }

        public async static Task<Connection> CreateAsync(string address)
        {
            var endPoint = EndPointFactory.Create(address);
            var socket = new Socket(AddressFamily.Unix, SocketType.Stream, 0);
            await socket.ConnectAsync(endPoint).ConfigureAwait(false);
            var stream = new NetworkStream(socket, ownsSocket: true);

            await authenticate(stream).ConfigureAwait(false);

            return new Connection(stream);
        }

        public IDisposable RegisterObjectProxy(
            ObjectPath path,
            string interfaceName,
            Func<uint, MessageHeader, byte[], Task> proxy
        )
        {
            var dictionaryEntry = path + "\0" + interfaceName;
            if (!objectProxies.TryAdd(dictionaryEntry, proxy))
                throw new InvalidOperationException("Attempted to register an object proxy twice");

            return new deregistration
            {
                Deregister = () =>
                {
                    Func<uint, MessageHeader, byte[], Task> _;
                    objectProxies.TryRemove(dictionaryEntry, out _);
                }
            };
        }

        public IDisposable RegisterSignalHandler(
            ObjectPath path,
            string interfaceName,
{'fraction_non_alphanumeric': 0.07645466847090664, 'fraction_numerical': 0.0033829499323410014, 'mean_word_length': 2.6010962241169304, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 35, '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': '2449578', 'n_tokens_mistral': 1412, 'n_tokens_neox': 1361, 'n_words': 398}