func New(
	beatName string,
	configs map[string]outputs.MothershipConfig,
	shipper ShipperConfig,
) (*PublisherType, error) {
	publisher := PublisherType{}
	err := publisher.init(beatName, configs, shipper)
	if err != nil {
		return nil, err
	}
	return &publisher, nil
}
Create new PublisherType
package publisher
import (
	"encoding/json"
	"errors"
	"flag"
	"os"
	"time"
	"github.com/elastic/beats/libbeat/common"
	"github.com/elastic/beats/libbeat/logp"
	"github.com/elastic/beats/libbeat/outputs"
	"github.com/nranchev/go-libGeoIP"
	_ "github.com/elastic/beats/libbeat/outputs/console"
	_ "github.com/elastic/beats/libbeat/outputs/elasticsearch"
	_ "github.com/elastic/beats/libbeat/outputs/fileout"
	_ "github.com/elastic/beats/libbeat/outputs/kafka"
	_ "github.com/elastic/beats/libbeat/outputs/logstash"
	_ "github.com/elastic/beats/libbeat/outputs/redis"
)
var publishDisabled *bool
var debug = logp.MakeDebug("publish")
type eventPublisher interface {
	PublishEvent(ctx Context, event common.MapStr) bool
	PublishEvents(ctx Context, events []common.MapStr) bool
}
type Context struct {
	publishOptions
	Signal outputs.Signaler
}
type publishOptions struct {
	Guaranteed bool
	Sync       bool
}
type TransactionalEventPublisher interface {
	PublishTransaction(transaction outputs.Signaler, events []common.MapStr)
}
type PublisherType struct {
	shipperName    string  
	hostname       string  
	name           string  
	IpAddrs        []string
	tags           []string
	disabled       bool
	Index          string
	Output         []*outputWorker
	TopologyOutput outputs.TopologyOutputer
	IgnoreOutgoing bool
	GeoLite        *libgeo.GeoIP
	RefreshTopologyTimer <-chan time.Time
	wsOutput    workerSignal
	wsPublisher workerSignal
	syncPublisher  *syncPublisher
	asyncPublisher *asyncPublisher
	client *client
}
type ShipperConfig struct {
	Name                  string
	Refresh_topology_freq int
	Ignore_outgoing       bool
	Topology_expire       int
	Tags                  []string
	Geoip                 common.Geoip
	QueueSize     *int `yaml:"queue_size"`
	BulkQueueSize *int `yaml:"bulk_queue_size"`
	MaxProcs *int `yaml:"max_procs"`
}
type Topology struct {
	Name string `json:"name"`
	Ip   string `json:"ip"`
}
const (
	defaultChanSize     = 1000
	defaultBulkChanSize = 0
)
func init() {
	publishDisabled = flag.Bool("N", false, "Disable actual publishing for testing")
}
func PrintPublishEvent(event common.MapStr) {
	json, err := json.MarshalIndent(event, "", "  ")
	if err != nil {
		logp.Err("json.Marshal: %s", err)
	} else {
		debug("Publish: %s", string(json))
	}
}
func (publisher *PublisherType) IsPublisherIP(ip string) bool {
	for _, myip := range publisher.IpAddrs {
		if myip == ip {
			return true
		}
	}
	return false
}
func (publisher *PublisherType) GetServerName(ip string) string {
	islocal, err := common.IsLoopback(ip)
	if err != nil {
		logp.Err("Parsing IP %s fails with: %s", ip, err)
		return ""
	}
	if islocal {
		return publisher.name
	}
	if publisher.TopologyOutput != nil {
		return publisher.TopologyOutput.GetNameByIP(ip)
	}
	return ""
}
func (publisher *PublisherType) Client() Client {
	return publisher.client
}
func (publisher *PublisherType) UpdateTopologyPeriodically() {
	for range publisher.RefreshTopologyTimer {
		_ = publisher.PublishTopology()  
	}
}
func (publisher *PublisherType) PublishTopology(params ...string) error {
	localAddrs := params
	if len(params) == 0 {
		addrs, err := common.LocalIpAddrsAsStrings(false)
		if err != nil {
			logp.Err("Getting local IP addresses fails with: %s", err)
			return err
		}
		localAddrs = addrs
	}
	if publisher.TopologyOutput != nil {
		debug("Add topology entry for %s: %s", publisher.name, localAddrs)
		err := publisher.TopologyOutput.PublishIPs(publisher.name, localAddrs)
		if err != nil {
			return err
		}
	}
	return nil
}