func NewPublisher(pub *publisher.Publisher, hwm, bulkHWM int) *PacketbeatPublisher {
	return &PacketbeatPublisher{
		pub:    pub,
		client: pub.Connect(),
		done:   make(chan struct{}),
		trans:  make(chan common.MapStr, hwm),
		flows:  make(chan []common.MapStr, bulkHWM),
	}
}

func NewPublisher(
	pub publisher.Publisher,
	hwm, bulkHWM int,
) (*PacketbeatPublisher, error) {
	topo, ok := pub.(TopologyProvider)
	if !ok {
		return nil, errors.New("Requires topology provider")
	}

	return &PacketbeatPublisher{
		pub:            pub,
		topo:           topo,
		geoLite:        topo.GeoLite(),
		ignoreOutgoing: topo.IgnoreOutgoing(),
		client:         pub.Connect(),
		done:           make(chan struct{}),
		trans:          make(chan common.MapStr, hwm),
		flows:          make(chan []common.MapStr, bulkHWM),
	}, nil
}

// newModuleWrappers creates new Modules and their associated MetricSets based
// on the given configuration. It constructs the supporting filters and
// publisher client and stores it all in a moduleWrapper.
func newModuleWrappers(
	modulesConfig []*common.Config,
	r *mb.Register,
	publisher *publisher.Publisher,
) ([]*moduleWrapper, error) {
	modules, err := mb.NewModules(modulesConfig, r)
	if err != nil {
		return nil, err
	}

	// Wrap the Modules and MetricSet's.
	var wrappers []*moduleWrapper
	var errs multierror.Errors
	for k, v := range modules {
		debugf("initializing Module type %s, %T=%+v", k.Name(), k, k)
		f, err := filter.New(k.Config().Filters)
		if err != nil {
			errs = append(errs, errors.Wrapf(err, "module %s", k.Name()))
			continue
		}

		mw := &moduleWrapper{
			Module:    k,
			filters:   f,
			pubClient: publisher.Connect(),
		}
		wrappers = append(wrappers, mw)

		msws := make([]*metricSetWrapper, 0, len(v))
		for _, ms := range v {
			debugf("initializing MetricSet type %s/%s, %T=%+v",
				ms.Module().Name(), ms.Name(), ms, ms)
			msw := &metricSetWrapper{
				MetricSet: ms,
				module:    mw,
				stats:     new(expvar.Map).Init(),
			}
			msws = append(msws, msw)

			// Initialize expvar stats for this MetricSet.
			fetches.Set(fmt.Sprintf("%s-%s", mw.Name(), msw.Name()), msw.stats)
			msw.stats.Add(successesKey, 0)
			msw.stats.Add(failuresKey, 0)
			msw.stats.Add(eventsKey, 0)
		}
		mw.metricSets = msws
	}

	return wrappers, errs.Err()
}

func normalizeTransAddr(pub *publisher.Publisher, event common.MapStr) bool {
	debugf("normalize address for: %v", event)

	var srcServer, dstServer string
	src, ok := event["src"].(*common.Endpoint)
	debugf("has src: %v", ok)
	if ok {
		// check if it's outgoing transaction (as client)
		isOutgoing := pub.IsPublisherIP(src.Ip)
		if isOutgoing {
			if pub.IgnoreOutgoing {
				// duplicated transaction -> ignore it
				debugf("Ignore duplicated transaction on: %s -> %s", srcServer, dstServer)
				return false
			}

			//outgoing transaction
			event["direction"] = "out"
		}

		srcServer = pub.GetServerName(src.Ip)
		event["client_ip"] = src.Ip
		event["client_port"] = src.Port
		event["client_proc"] = src.Proc
		event["client_server"] = srcServer
		delete(event, "src")
	}

	dst, ok := event["dst"].(*common.Endpoint)
	debugf("has dst: %v", ok)
	if ok {
		dstServer = pub.GetServerName(dst.Ip)
		event["ip"] = dst.Ip
		event["port"] = dst.Port
		event["proc"] = dst.Proc
		event["server"] = dstServer
		delete(event, "dst")

		//check if it's incoming transaction (as server)
		if pub.IsPublisherIP(dst.Ip) {
			// incoming transaction
			event["direction"] = "in"
		}

	}

	if pub.GeoLite != nil {
		realIP, exists := event["real_ip"]
		if exists && len(realIP.(common.NetString)) > 0 {
			loc := pub.GeoLite.GetLocationByIP(string(realIP.(common.NetString)))
			if loc != nil && loc.Latitude != 0 && loc.Longitude != 0 {
				loc := fmt.Sprintf("%f, %f", loc.Latitude, loc.Longitude)
				event["client_location"] = loc
			}
		} else {
			if len(srcServer) == 0 && src != nil { // only for external IP addresses
				loc := pub.GeoLite.GetLocationByIP(src.Ip)
				if loc != nil && loc.Latitude != 0 && loc.Longitude != 0 {
					loc := fmt.Sprintf("%f, %f", loc.Latitude, loc.Longitude)
					event["client_location"] = loc
				}
			}
		}
	}

	return true
}