<commit_before>//
// daemon.go
//
package main

import (
  "bufio"
  "log"
  "net"
  "strconv"
  "strings"
  "time"
)

type NNTPDaemon struct {
  instance_name string
  bind_addr string
  conf *SRNdConfig
  store *ArticleStore
  api_caller *API
  listener net.Listener
  debug bool
  sync_on_start bool
  running bool
  feeds map[NNTPConnection]bool
  infeed chan string
}

func (self *NNTPDaemon) End() {
  self.listener.Close()
}

// register a new connection
// can be either inbound or outbound
func (self *NNTPDaemon) newConnection(conn net.Conn, inbound bool, policy *FeedPolicy) NNTPConnection {
  feed := NNTPConnection{conn, bufio.NewReader(conn), inbound, self.debug, new(ConnectionInfo), policy, make(chan *NNTPMessage)}
  self.feeds[feed] = ! inbound
  return feed
}

func (self *NNTPDaemon) persistFeed(conf FeedConfig) {
  for {
    if self.running {
      time.Sleep(1 * time.Second)
      var conn net.Conn
      var err error
      proxy_type := strings.ToLower(conf.proxy_type)
      
      if proxy_type ==  "" || proxy_type == "none" {
        // connect out without proxy 
        log.Println("dial out to ", conf.addr)
        conn, err = net.Dial("tcp", conf.addr)
        if err != nil {
          log.Println("cannot connect to outfeed", conf.addr, err)
          continue
        }
      } else if proxy_type == "socks4a" {
        // connect via socks4a
        log.Println("dial out via proxy", conf.proxy_addr)
        conn, err = net.Dial("tcp", conf.proxy_addr)
        if err != nil {
          log.Println("cannot connect to proxy", conf.proxy_addr)
          continue
        }
        // generate request
        idx := strings.LastIndex(conf.addr, ":")
        if idx == -1 {
          log.Fatal("invalid outfeed address")
        }
        var port uint64
        addr := conf.addr[:idx]
        port, err = strconv.ParseUint(conf.addr[idx+1:], 10, 16)
        if port >= 25536 {
          log.Fatal("bad proxy port" , port)
        }
        var proxy_port uint16
        proxy_port = uint16(port)
        proxy_ident := "sr<PASSWORD>"
        req_len := len(addr) + 1 + len(proxy_ident) + 1 + 8

        req := make([]byte, req_len)
        // pack request
        req[0] = '\x04'
        req[1] = '\x01'
        req[2] = byte(proxy_port & 0xff00 >> 8)
        req[3] = byte(proxy_port & 0x00ff)
        req[7] = '\x01'
        idx = 8
        
        proxy_ident_b := []byte(proxy_ident)
        addr_b := []byte(addr)
        
        var bi int
        for bi = range proxy_ident_b {
          req[idx] = proxy_ident_b[bi]
          idx += 1
        }
        idx += 1
        for bi = range addr_b {
          req[idx] = addr_b[bi]
          idx += 1
        }
  
        // send request
        conn.Write(req)
        resp := make([]byte, 8)
        
        // receive response
        conn.Read(resp)
        if resp[1] == '\x5a' {
          // success
          log.Println("connected to", conf.addr)
        } else {
          log.Println("failed to connect to", conf.addr)
          continue
        }
      }
      policy := &conf.policy
      nntp := self.newConnection(conn, false, policy)
      nntp.HandleOutbound(self)
      delete(self.feeds, nntp)
    }
  }
}

// run daemon
func (self *NNTPDaemon) Run() {	
  err := self.Bind()
  if err != nil {
    log.Println("failed to bind:", err)
    return
  }
  defer self.listener.Close()

  // we are now running
  self.running = true
  
  // persist outfeeds
  for idx := range self.conf.feeds {
    go self.persistFeed(self.conf.feeds[idx])
  }
  go self.mainloop()
  
  if self.sync_on_start {
    log.Println("syncing on start")
    go self.syncAll()
  }
  
  // loop over messages
  for {
    message := <- self.infeed
    // load message
    nntp := self.store.GetMessage(message, false)
    // send to all outfeeds
    for feed , use := range self.feeds {
      if use {
        feed.send <- nntp
      }
    }
  }
}

func (self *NNTPDaemon) syncAll() {
  self.store.IterateAllArticles(func(messageID string) {
    msg := self.store.GetMessage(messageID, false)
    for feed , use := range self.feeds {
      if use {
        feed.send <- msg
     }
    }
  }) 
}

func (self *NNTPDaemon) mainloop() {	
  for {
    // accept
    conn, err := self.listener.Accept()
    if err != nil {
      log.Fatal(err)
    }
    // make a new inbound nntp connection handler 
    nntp := self.newConnection(conn, true, nil)
    go self.RunConnection(nntp)
  }
}

func (self *NNTPDaemon) RunConnection(nntp NNTPConnection) {
  nntp.HandleInbound(self)
  delete(self.feeds, nntp)
}

// bind to address
func (self *NNTPDaemon) Bind() error {<commit_msg>fix outfeed sync on start<commit_after>//
// daemon.go
//
package main

import (
  "bufio"
  "log"
  "net"
  "strconv"
  "strings"
  "time"
)

type NNTPDaemon struct {
  instance_name string
  bind_addr string
  conf *SRNdConfig
  store *ArticleStore
  api_caller *API
  listener net.Listener
  debug bool
  sync_on_start bool
  running bool
  feeds map[NNTPConnection]bool
  infeed chan string
}

func (self *NNTPDaemon) End() {
  self.listener.Close()
}

// register a new connection
// can be either inbound or outbound
func (self *NNTPDaemon) newConnection(conn net.Conn, inbound bool, policy *FeedPolicy) NNTPConnection {
  feed := NNTPConnection{conn, bufio.NewReader(conn), inbound, self.debug, new(ConnectionInfo), policy, make(chan *NNTPMessage)}
  self.feeds[feed] = ! inbound
  return feed
}

func (self *NNTPDaemon) persistFeed(conf FeedConfig) {
  for {
    if self.running {
      time.Sleep(1 * time.Second)
      var conn net.Conn
      var err error
      proxy_type := strings.ToLower(conf.proxy_type)
      
      if proxy_type ==  "" || proxy_type == "none" {
        // connect out without proxy 
        log.Println("dial out to ", conf.addr)
        conn, err = net.Dial("tcp", conf.addr)
        if err != nil {
          log.Println("cannot connect to outfeed", conf.addr, err)
          continue
        }
      } else if proxy_type == "socks4a" {
        // connect via socks4a
        log.Println("dial out via proxy", conf.proxy_addr)
        conn, err = net.Dial("tcp", conf.proxy_addr)
        if err != nil {
          log.Println("cannot connect to proxy", conf.proxy_addr)
          continue
        }
        // generate request
        idx := strings.LastIndex(conf.addr, ":")
        if idx == -1 {
          log.Fatal("invalid outfeed address")
        }
        var port uint64
        addr := conf.addr[:idx]
        port, err = strconv.ParseUint(conf.addr[idx+1:], 10, 16)
        if port >= 25536 {
          log.Fatal("bad proxy port" , port)
        }
        var proxy_port uint16
        proxy_port = uint16(port)
        proxy_ident := "srndv2"
        req_len := len(addr) + 1 + len(proxy_ident) + 1 + 8

        req := make([]byte, req_len)
        // pack request
        req[0] = '\x04'
        req[1] = '\x01'
        req[2] = byte(proxy_port & 0xff00 >> 8)
        req[3] = byte(proxy_port & 0x00ff)
        req[7] = '\x01'
        idx = 8
        
        proxy_ident_b := []byte(proxy_ident)
        addr_b := []byte(addr)
        
        var bi int
        for bi = range proxy_ident_b {
          req[idx] = proxy_ident_b[bi]
          idx += 1
        }
        idx += 1
        for bi = range addr_b {
          req[idx] = addr_b[bi]
          idx += 1
        }
  
        // send request
        conn.Write(req)
        resp := make([]byte, 8)
        
        // receive response
        conn.Read(resp)
        if resp[1] == '\x5a' {
          // success
          log.Println("connected to", conf.addr)
        } else {
          log.Println("failed to connect to", conf.addr)
          continue
        }
      }
      policy := &conf.policy
      nntp := self.newConnection(conn, false, policy)
      nntp.HandleOutbound(self)
      delete(self.feeds, nntp)
    }
  }
}

// run daemon
func (self *NNTPDaemon) Run() {	
  err := self.Bind()
  if err != nil {
    log.Println("failed to bind:", err)
    return
  }
  defer self.listener.Close()

  // we are now running
  self.running = true
  
  // persist outfeeds
  for idx := range self.conf.feeds {
    go self.persistFeed(self.conf.feeds[idx])
  }
  go self.mainloop()
  
  // loop over messages
  for {
    message := <- self.infeed
    // load message
    nntp := self.store.GetMessage(message, false)
    // send to all outfeeds
    for feed , use := range self.feeds {
      if use {
        feed.send <- nntp
      }
    }
  }
}

func (self *NNTPDaemon) mainloop() {	
  for {
    // accept
    conn, err := self.listener.Accept()
    if err != nil {
      log.Fatal(err)
    }
    // make a new inbound nntp connection handler 
    nntp := self.newConnection(conn, true, nil)
    go self.RunConnection(nntp)
  }
}

func (self *NNTPDaemon) RunConnection(nntp NNTPConnection) {
  nntp.HandleInbound(self)
  delete(self.feeds, nntp)
}

// bind to address
func (self *NNTPDaemon) Bind() error {
{'fraction_non_alphanumeric': 0.12318678866324481, 'fraction_numerical': 0.010823476902477126, 'mean_word_length': 2.0137861466039007, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 10, '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': '1299640', 'n_tokens_mistral': 3100, 'n_tokens_neox': 3023, 'n_words': 751}