<commit_before>package stack

import "log"
import "io/ioutil"

import "crypto/x509"
import "encoding/pem"
import "encoding/json"

import tlvCodec "github.com/chris-wood/spud/codec"
import "github.com/chris-wood/spud/tables/lpm"
import "github.com/chris-wood/spud/messages"
import "github.com/chris-wood/spud/messages/name"
import "github.com/chris-wood/spud/stack/config"
import "github.com/chris-wood/spud/stack/cache"
import "github.com/chris-wood/spud/stack/pit"
import "github.com/chris-wood/spud/stack/component/tunnel"
import "github.com/chris-wood/spud/stack/component/connector"
import "github.com/chris-wood/spud/stack/component/codec"
import "github.com/chris-wood/spud/stack/component/crypto"
import "github.com/chris-wood/spud/stack/component/crypto/validator"
import "github.com/chris-wood/spud/stack/component/crypto/context"

type MessageCallback func(msg messages.MessageWrapper)

type Component interface {
    // XXX: rename to Push and Pop, respectively
    Enqueue(messages.MessageWrapper)
    Dequeue() messages.MessageWrapper
    ProcessEgressMessages()
    ProcessIngressMessages()
}

type Stack struct {
    cryptoComponent Component
    codecComponent Component
    head Component
    bottom Component

    pendingMap map[string]MessageCallback
    pendingQueue chan messages.MessageWrapper
    prefixTable lpm.LPM
}

func (s *Stack) Enqueue(msg messages.MessageWrapper) {
    s.head.Enqueue(msg)
}

func (s *Stack) Dequeue() messages.MessageWrapper {
    return <- s.pendingQueue
}

func (s *Stack) Get(msg messages.MessageWrapper, callback MessageCallback) {
    if (msg.GetPacketType() != tlvCodec.T_INTEREST) {
        return
    }

    // Register the callback
    s.pendingMap[msg.Identifier()] = callback

    // Enqueue the message into the top of the stack
    s.head.Enqueue(msg)
}

func (s *Stack) Service(nameString string, callback MessageCallback) {
    prefixName, err := name.Parse(nameString)
    if err == nil {
        nameComponents := prefixName.SegmentStrings()
        s.prefixTable.Insert(nameComponents, callback)
    }
}

func (s *Stack) processInputQueue() {
    for ;; {
        // Dequeue messages as they arrive
        msg := s.head.Dequeue()

        switch (msg.GetPacketType()) {
        case tlvCodec.T_INTEREST:
            requestName := msg.Name()
            numSsegments := len(requestName.Segments)

            handled := false
            for index := 1; index <= numSsegments; index++ {
                nameComponents := requestName.SegmentStrings()
                callbackInterface, ok := s.prefixTable.Lookup(nameComponents)

                if ok {
                    callback := callbackInterface.(MessageCallback)
                    callback(msg)
                    s.prefixTable.Drop(nameComponents)
                    handled = true
                }
            }

            if !handled {
                s.pendingQueue <- msg
            }
        default:
            callback, ok := s.pendingMap[msg.Identifier()]
            if ok {
                callback(msg)
            } else {
                s.pendingQueue <- msg
            }
        }
    }
}

func (s *Stack) AddTunnel(tun tunnel.TunnelComponent) {
}

func Create(config config.StackConfig) (*Stack, error) {
  var err error

  // Create the link
  var fc connector.ForwarderConnector
  switch config.Link {
  case "tcp":
      locator := config.FwdAddress
      fc, err = connector.NewAthenaTCPForwarderConnector(locator)
      if err != nil {
          panic(err)
      }
      break
  case "loopback":
      fallthrough
  default:
      fc, _ = connector.NewLoopbackForwarderConnector()
      break
  }

  if fc == nil {
      log.Panic("Catastrophic failure: a connector was not created.")
  }

  // Create the shared data structures
  stackCache := cache.NewCache()
  stackPit := pit.NewPIT()

  // Create codec
  codecComponent := codec.NewCodecComponent(fc, stackCache, stackPit)
  go codecComponent.ProcessEgressMessages()
  go codecComponent.ProcessIngressMessages()

  // Create crypto component
  // XXX: defer validator and encryptor creation to the crypto component
  trustStore := context.NewTrustStore()
  cryptoComponent := crypto.NewCryptoComponent(trustStore, codecComponent)
  go cryptoComponent.ProcessEgressMessages()
  go cryptoComponent.ProcessIngressMessages()

  // Create the right crypto processors
  if len(config.Keys) > 0 {
      for _, keyFileName := range(config.Keys) {
          keyData, err := ioutil.ReadFile(keyFileName)
          block, _ := pem.Decode(keyData)
          privateKey, parseError := x509.ParsePKCS1PrivateKey(block.Bytes)
          if parseError != nil {
              log.Printf("Failed to parse private key: %s", err)
          } else {
              rsaProcessor, _ := validator.NewRSAProcessorWithKey(privateKey)
              cryptoComponent.AddCryptoProcessor("/", rsaProcessor)
          }
      }
  } else {
      rsaProcessor, _ := validator.NewRSAProcessor(2048)
      cryptoComponent.AddCryptoProcessor("/", rsaProcessor)
  }

  // Assemble the stack
  stack := &Stack{
      cryptoComponent: cryptoComponent,
      codecComponent: codecComponent,
      head: cryptoComponent,
      pendingQueue: make(chan messages.MessageWrapper, config.PendingBufferSize), // random constant -- make this configurable
      pendingMap: make(map[string]MessageCallback),
      prefixTable: lpm.LPM{},
  }

  // Start the stack processing loop
  go stack.processInputQueue()

  return stack, nil
}<commit_msg>Defer component initialization to the very end<commit_after>package stack

import "log"
import "io/ioutil"

import "crypto/x509"
import "encoding/pem"
import "encoding/json"

import tlvCodec "github.com/chris-wood/spud/codec"
import "github.com/chris-wood/spud/tables/lpm"
import "github.com/chris-wood/spud/messages"
import "github.com/chris-wood/spud/messages/name"
import "github.com/chris-wood/spud/stack/config"
import "github.com/chris-wood/spud/stack/cache"
import "github.com/chris-wood/spud/stack/pit"
import "github.com/chris-wood/spud/stack/component/tunnel"
import "github.com/chris-wood/spud/stack/component/connector"
import "github.com/chris-wood/spud/stack/component/codec"
import "github.com/chris-wood/spud/stack/component/crypto"
import "github.com/chris-wood/spud/stack/component/crypto/validator"
import "github.com/chris-wood/spud/stack/component/crypto/context"

type MessageCallback func(msg messages.MessageWrapper)

type Component interface {
    // XXX: rename to Push and Pop, respectively
    Enqueue(messages.MessageWrapper)
    Dequeue() messages.MessageWrapper
    ProcessEgressMessages()
    ProcessIngressMessages()
}

type Stack struct {
    cryptoComponent Component
    codecComponent Component
    head Component
    bottom Component

    pendingMap map[string]MessageCallback
    pendingQueue chan messages.MessageWrapper
    prefixTable lpm.LPM
}

func (s *Stack) Enqueue(msg messages.MessageWrapper) {
    s.head.Enqueue(msg)
}

func (s *Stack) Dequeue() messages.MessageWrapper {
    return <- s.pendingQueue
}

func (s *Stack) Get(msg messages.MessageWrapper, callback MessageCallback) {
    if (msg.GetPacketType() != tlvCodec.T_INTEREST) {
        return
    }

    // Register the callback
    s.pendingMap[msg.Identifier()] = callback

    // Enqueue the message into the top of the stack
    s.head.Enqueue(msg)
}

func (s *Stack) Service(nameString string, callback MessageCallback) {
    prefixName, err := name.Parse(nameString)
    if err == nil {
        nameComponents := prefixName.SegmentStrings()
        s.prefixTable.Insert(nameComponents, callback)
    }
}

func (s *Stack) processInputQueue() {
    for ;; {
        // Dequeue messages as they arrive
        msg := s.head.Dequeue()

        switch (msg.GetPacketType()) {
        case tlvCodec.T_INTEREST:
            requestName := msg.Name()
            numSsegments := len(requestName.Segments)

            handled := false
            for index := 1; index <= numSsegments; index++ {
                nameComponents := requestName.SegmentStrings()
                callbackInterface, ok := s.prefixTable.Lookup(nameComponents)

                if ok {
                    callback := callbackInterface.(MessageCallback)
                    callback(msg)
                    s.prefixTable.Drop(nameComponents)
                    handled = true
                }
            }

            if !handled {
                s.pendingQueue <- msg
            }
        default:
            callback, ok := s.pendingMap[msg.Identifier()]
            if ok {
                callback(msg)
            } else {
                s.pendingQueue <- msg
            }
        }
    }
}

func (s *Stack) AddTunnel(tun tunnel.TunnelComponent) {
}

func Create(config config.StackConfig) (*Stack, error) {
  var err error

  // Create the link
  var fc connector.ForwarderConnector
  switch config.Link {
  case "tcp":
      locator := config.FwdAddress
      fc, err = connector.NewAthenaTCPForwarderConnector(locator)
      if err != nil {
          panic(err)
      }
      break
  case "loopback":
      fallthrough
  default:
      fc, _ = connector.NewLoopbackForwarderConnector()
      break
  }

  if fc == nil {
      log.Panic("Catastrophic failure: a connector was not created.")
  }

  // Create the shared data structures
  stackCache := cache.NewCache()
  stackPit := pit.NewPIT()

  // Create codec
  codecComponent := codec.NewCodecComponent(fc, stackCache, stackPit)

  // Create crypto component
  // XXX: defer validator and encryptor creation to the crypto component
  trustStore := context.NewTrustStore()
  cryptoComponent := crypto.NewCryptoComponent(trustStore, codecComponent)

  // Create the right crypto processors
  if len(config.Keys) > 0 {
      for _, keyFileName := range(config.Keys) {
          keyData, err := ioutil.ReadFile(keyFileName)
          block, _ := pem.Decode(keyData)
          privateKey, parseError := x509.ParsePKCS1PrivateKey(block.Bytes)
          if parseError != nil {
              log.Printf("Failed to parse private key: %s", err)
          } else {
              rsaProcessor, _ := validator.NewRSAProcessorWithKey(privateKey)
              cryptoComponent.AddCryptoProcessor("/", rsaProcessor)
          }
      }
  } else {
      rsaProcessor, _ := validator.NewRSAProcessor(2048)
      cryptoComponent.AddCryptoProcessor("/", rsaProcessor)
  }

  // Assemble the stack
  stack := &Stack{
      cryptoComponent: cryptoComponent,
      codecComponent: codecComponent,
      head: cryptoComponent,
      pendingQueue: make(chan messages.MessageWrapper, config.PendingBufferSize), // random constant -- make this configurable
      pendingMap: make(map[string]MessageCallback),
      prefixTable: lpm.LPM{},
  }

  // Start the stack processing loop
  go codecComponent.ProcessEgressMessages()
  go codecComponent.ProcessIngressMessages()
  go cryptoComponent.ProcessEgressMessages()
  go cryptoComponent.ProcessIngressMessages()
  go stack.processInputQueue()

  return stack, nil
}
{'fraction_non_alphanumeric': 0.10229021453788359, 'fraction_numerical': 0.002353580157508826, 'mean_word_length': 2.7514431239388792, 'pattern_counts': {'":': 4, '<': 11, '<?xml version=': 0, '>': 5, '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': '295628', 'n_tokens_mistral': 3272, 'n_tokens_neox': 3152, 'n_words': 708}