<commit_before>﻿module SwimProtocol.Swim

open System
open Membership
open Transport
open FailureDetection

type Host = string * int

type Config = 
    { Port: int
      PeriodTimeout : TimeSpan
      PingTimeout: TimeSpan
      PingRequestGroupSize : int
      SuspectTimeout : TimeSpan }

let defaultConfig =
    { Port = 1337
      PeriodTimeout = TimeSpan.FromSeconds 2.0
      PingTimeout = TimeSpan.FromMilliseconds 200.0
      PingRequestGroupSize = 3
      SuspectTimeout = TimeSpan.FromSeconds 10.0 }

let private decodeMessage (addr, bytes) =
    Message.decode bytes
    |> Option.map (fun msg -> addr, msg)

let init config hosts = 
    let { Port = port
          PeriodTimeout = periodTimeout
          PingTimeout = pingTimeout
          PingRequestGroupSize = pingReqGrpSize
          SuspectTimeout = suspectTimeout } = config

    let local = MemberList.makeLocal port
    let disseminator = Dissemination.create()
    let udp = Udp.connect port
    
    let memberList =
        hosts
        |> List.map (fun (h, p) -> MemberList.makeMember h p)
        |> MemberList.createWith disseminator suspectTimeout

    let detectorConfig = { Local = local
                           MemberList = memberList
                           PeriodTimeout = periodTimeout
                           PingTimeout = pingTimeout
                           PingRequestGroupSize = pingReqGrpSize }

    udp.Received
    |> Observable.choose decodeMessage
    |> FailureDetection.run detectorConfig
    |> Observable.map (fun (addr, msg) -> addr, Message.encode msg)
    |> Observable.subscribe udp.Send<commit_msg>reorganize code<commit_after>﻿module SwimProtocol.Swim

open System
open Membership
open Transport
open FailureDetection

type Host = string * int

type Config = 
    { Port: int
      PeriodTimeout : TimeSpan
      PingTimeout: TimeSpan
      PingRequestGroupSize : int
      SuspectTimeout : TimeSpan }

let defaultConfig =
    { Port = 1337
      PeriodTimeout = TimeSpan.FromSeconds 2.0
      PingTimeout = TimeSpan.FromMilliseconds 200.0
      PingRequestGroupSize = 3
      SuspectTimeout = TimeSpan.FromSeconds 10.0 }

let private decodeMessage (addr, bytes) =
    Message.decode bytes
    |> Option.map (fun msg -> addr, msg)

let init config hosts = 
    let { Port = port
          PeriodTimeout = periodTimeout
          PingTimeout = pingTimeout
          PingRequestGroupSize = pingReqGrpSize
          SuspectTimeout = suspectTimeout } = config

    let local = MemberList.makeLocal port
    let disseminator = Dissemination.create()
    let udp = Udp.connect port
    
    let memberList =
        hosts
        |> List.map (fun (h, p) -> MemberList.makeMember h p)
        |> MemberList.createWith disseminator suspectTimeout

    udp.Received
    |> Observable.choose decodeMessage
    // dispach event to disseminator
    |> FailureDetection.run { Local = local
                              MemberList = memberList
                              PeriodTimeout = periodTimeout
                              PingTimeout = pingTimeout
                              PingRequestGroupSize = pingReqGrpSize }
    |> Observable.map (fun (addr, msg) -> addr, Message.encode msg)
    // Add event from disseminator
    |> Observable.subscribe udp.Send
{'fraction_non_alphanumeric': 0.06507791017415215, 'fraction_numerical': 0.008554842652001222, 'mean_word_length': 2.2707292707292708, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, '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': '482387', 'n_tokens_mistral': 914, 'n_tokens_neox': 848, 'n_words': 243}