<commit_before>package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/garyburd/redigo/redis"
)

type RedisDataType uint16

const (
	RedisListType RedisDataType = iota
	RedisChannelType
)

type RedisOutputType struct {
	OutputInterface
	Index string
	Conn  redis.Conn

	TopologyExpire     time.Duration
	ReconnectInterval  time.Duration
	Hostname           string
	Password           string
	Db                 int
	DbTopology         int
	Timeout            time.Duration
	DataType           RedisDataType
	FlushInterval      time.Duration
	flush_immediatelly bool

	TopologyMap  map[string]string
	sendingQueue chan RedisQueueMsg
	connected    bool
}

type RedisQueueMsg struct {
	index string
	msg   string
}

var RedisOutput RedisOutputType

func (out *RedisOutputType) Init(config tomlMothership) error {

	out.Hostname = fmt.Sprintf("%s:%d", config.Host, config.Port)

	if config.Password != "" {
		out.Password = config.Password
	}

	if config.Db != 0 {
		out.Db = config.Db
	}

	out.DbTopology = 1
	if config.Db_topology != 0 {
		out.DbTopology = config.Db_topology
	}

	out.Timeout = 5 * time.Second
	if config.Timeout != 0 {
		out.Timeout = time.Duration(config.Timeout) * time.Second
	}

	if config.Index != "" {
		out.Index = config.Index
	} else {
		out.Index = "packetbeat"
	}

	out.FlushInterval = 1000 * time.Millisecond
	if config.Flush_interval != 0 {
		if config.Flush_interval < 0 {
			out.flush_immediatelly = true
			WARN("Flushing to REDIS on each push, performance migh be affected")
		} else {
			out.FlushInterval = out.FlushInterval * time.Millisecond
		}
	}

	out.ReconnectInterval = time.Duration(1) * time.Second
	if config.Reconnect_interval != 0 {
		out.ReconnectInterval = time.Duration(config.Reconnect_interval) * time.Second
	}

	exp_sec := 15
	if _Config.Agent.Topology_expire != 0 {
		exp_sec = _Config.Agent.Topology_expire
	}
	out.TopologyExpire = time.Duration(exp_sec) * time.Second

	switch config.DataType {
	case "", "list":
		out.DataType = RedisListType
	case "channel":
		out.DataType = RedisChannelType
	default:
		return errors.New("Bad Redis data type")
	}

	INFO("[RedisOutput] Using Redis server %s", out.Hostname)
	if out.Password != "" {
		INFO("[RedisOutput] Using password to connect to Redis")
	}
	INFO("[RedisOutput] Redis connection timeout %s", out.Timeout)
	INFO("[RedisOutput] Redis reconnect interval %s", out.ReconnectInterval)
	INFO("[RedisOutput] Using index pattern %s", out.Index)
	INFO("[RedisOutput] Topology expires after %s", out.TopologyExpire)
	INFO("[RedisOutput] Using db %d for storing events", out.Db)
	INFO("[RedisOutput] Using db %d for storing topology", out.DbTopology)
	INFO("[RedisOutput] Using %d data type", out.DataType)

	out.sendingQueue = make(chan RedisQueueMsg, 1000)

	out.Reconnect()
	go out.SendMessagesGoroutine()

	return nil
}

func (out *RedisOutputType) RedisConnect(db int) (redis.Conn, error) {
	conn, err := redis.DialTimeout(
		"tcp",
		out.Hostname,
		out.Timeout, out.Timeout, out.Timeout)
	if err != nil {
		return nil, err<commit_msg>Fixed flush_interval configuration<commit_after>package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/garyburd/redigo/redis"
)

type RedisDataType uint16

const (
	RedisListType RedisDataType = iota
	RedisChannelType
)

type RedisOutputType struct {
	OutputInterface
	Index string
	Conn  redis.Conn

	TopologyExpire     time.Duration
	ReconnectInterval  time.Duration
	Hostname           string
	Password           string
	Db                 int
	DbTopology         int
	Timeout            time.Duration
	DataType           RedisDataType
	FlushInterval      time.Duration
	flush_immediatelly bool

	TopologyMap  map[string]string
	sendingQueue chan RedisQueueMsg
	connected    bool
}

type RedisQueueMsg struct {
	index string
	msg   string
}

var RedisOutput RedisOutputType

func (out *RedisOutputType) Init(config tomlMothership) error {

	out.Hostname = fmt.Sprintf("%s:%d", config.Host, config.Port)

	if config.Password != "" {
		out.Password = config.Password
	}

	if config.Db != 0 {
		out.Db = config.Db
	}

	out.DbTopology = 1
	if config.Db_topology != 0 {
		out.DbTopology = config.Db_topology
	}

	out.Timeout = 5 * time.Second
	if config.Timeout != 0 {
		out.Timeout = time.Duration(config.Timeout) * time.Second
	}

	if config.Index != "" {
		out.Index = config.Index
	} else {
		out.Index = "packetbeat"
	}

	out.FlushInterval = 1000 * time.Millisecond
	if config.Flush_interval != 0 {
		if config.Flush_interval < 0 {
			out.flush_immediatelly = true
			WARN("Flushing to REDIS on each push, performance migh be affected")
		} else {
			out.FlushInterval = time.Duration(config.Flush_interval) * time.Millisecond
		}
	}

	out.ReconnectInterval = time.Duration(1) * time.Second
	if config.Reconnect_interval != 0 {
		out.ReconnectInterval = time.Duration(config.Reconnect_interval) * time.Second
	}

	exp_sec := 15
	if _Config.Agent.Topology_expire != 0 {
		exp_sec = _Config.Agent.Topology_expire
	}
	out.TopologyExpire = time.Duration(exp_sec) * time.Second

	switch config.DataType {
	case "", "list":
		out.DataType = RedisListType
	case "channel":
		out.DataType = RedisChannelType
	default:
		return errors.New("Bad Redis data type")
	}

	INFO("[RedisOutput] Using Redis server %s", out.Hostname)
	if out.Password != "" {
		INFO("[RedisOutput] Using password to connect to Redis")
	}
	INFO("[RedisOutput] Redis connection timeout %s", out.Timeout)
	INFO("[RedisOutput] Redis reconnect interval %s", out.ReconnectInterval)
	INFO("[RedisOutput] Redis flushing interval %s", out.FlushInterval)
	INFO("[RedisOutput] Using index pattern %s", out.Index)
	INFO("[RedisOutput] Topology expires after %s", out.TopologyExpire)
	INFO("[RedisOutput] Using db %d for storing events", out.Db)
	INFO("[RedisOutput] Using db %d for storing topology", out.DbTopology)
	INFO("[RedisOutput] Using %d data type", out.DataType)

	out.sendingQueue = make(chan RedisQueueMsg, 1000)

	out.Reconnect()
	go out.SendMessagesGoroutine()

	return nil
}

func (out *RedisOutputType) RedisConnect(db int) (redis.Conn, error) {
	conn, err := redis.DialTimeout(
		"tcp",
		out.Hostname,
		out.Timeout, out.Timeout, out.Timeout)
	if err != nil {
		return nil, err
{'fraction_non_alphanumeric': 0.10761615839054767, 'fraction_numerical': 0.007025387194635159, 'mean_word_length': 3.8408037094281298, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 3, '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': '2920013', 'n_tokens_mistral': 2247, 'n_tokens_neox': 2143, 'n_words': 517}