// OnData builds a composed decider that fires on when the PredicateFunc is satisfied.
491168
func OnData(predicate PredicateFunc, deciders ...Decider) Decider {
	return func(ctx *FSMContext, h *swf.HistoryEvent, data interface{}) Outcome {
		if predicate(data) {
			logf(ctx, "at=on-data")
			return NewComposedDecider(deciders...)(ctx, h, data)
		}
		return ctx.Pass()
	}
}
106356
func OnChildStarted(deciders ...Decider) Decider {
	return func(ctx *FSMContext, h *swf.HistoryEvent, data interface{}) Outcome {
		switch *h.EventType {
		case swf.EventTypeChildWorkflowExecutionStarted:
			logf(ctx, "at=on-child-started")
			return NewComposedDecider(deciders...)(ctx, h, data)
		}
		return ctx.Pass()
	}
}
579596
func (c *Context) BufferData(target int, data interface{}, usage int) {
	c.Call("bufferData", target, data, usage)
}
481507
func (s *APNSMessage) SetData(v map[string]*string) *APNSMessage {
	s.Data = v
	return s
}
28177
func (c *Callbacks) OnCDemoCustomData(fn func(*dota.CDemoCustomData) error) {
	c.onCDemoCustomData = append(c.onCDemoCustomData, fn)
}
1183432
func (s *RawEmail) SetData(v []byte) *RawEmail {
	s.Data = v
	return s
}
715304
func (s *ThirdPartyJobDetails) SetData(v *ThirdPartyJobData) *ThirdPartyJobDetails {
	s.Data = v
	return s
}
653229
func (p *Plot) SetData(data table.Grouping) *Plot {
	p.env.data = data
	return p
}
708632
func (s *ADMMessage) SetData(v map[string]*string) *ADMMessage {
	s.Data = v
	return s
}
1122681
func (s *InventoryResultEntity) SetData(v map[string]*InventoryResultItem) *InventoryResultEntity {
	s.Data = v
	return s
}
1095301
func (s *PutRecordInput) SetData(v []byte) *PutRecordInput {
	s.Data = v
	return s
}
1082622
func (e *Execution) DataProto() (ret *dm.Execution_Data) {
	switch e.State {
	case dm.Execution_SCHEDULING:
		ret = dm.NewExecutionScheduling().Data
	case dm.Execution_RUNNING:
		ret = dm.NewExecutionRunning().Data
	case dm.Execution_STOPPING:
		ret = dm.NewExecutionStopping().Data
	case dm.Execution_FINISHED:
		ret = dm.NewExecutionFinished(e.Result.Data).Data
	case dm.Execution_ABNORMAL_FINISHED:
		ret = dm.NewExecutionAbnormalFinish(e.Result.AbnormalFinish).Data
	default:
		panic(fmt.Errorf("unknown Execution_State: %s", e.State))
	}
	ret.Created = google_pb.NewTimestamp(e.Created)
	ret.Modified = google_pb.NewTimestamp(e.Modified)
	ret.DistributorInfo = &dm.Execution_Data_DistributorInfo{
		ConfigName:    e.DistributorConfigName,
		ConfigVersion: e.DistributorConfigVersion,
		Token:         e.DistributorToken,
	}
	return ret
}
446955
func CallNoData(ctx context.Context, caller Caller, call *scheduler.Call) error {
	resp, err := caller.Call(ctx, call)
	if resp != nil {
		resp.Close()
	}
	return err
}
140995
func (c *Callbacks) OnCDOTAMatchMetadataFile(fn func(*dota.CDOTAMatchMetadataFile) error) {
	c.onCDOTAMatchMetadataFile = append(c.onCDOTAMatchMetadataFile, fn)
}
846428
func (r *Request) SetData(data []byte) {
	r.data = data
	r.Signature = nil
}
753235
func (s *Certificate) SetData(v string) *Certificate {
	s.Data = &v
	return s
}
1229126
func (p *Parser) OnData(reply, endStream bool, dataBuffers [][]byte) (proxylib.OpType, int) {
	if reply {
		if p.injectFromQueue() {
			return proxylib.INJECT, len(DeniedMsgBase)
		}
		if len(dataBuffers) == 0 {
			return proxylib.NOP, 0
		}
	}

	//TODO don't copy data from buffers
	data := bytes.Join(dataBuffers, []byte{})
	log.Debugf("Data length: %d", len(data))

	if headerSize > len(data) {
		headerMissing := headerSize - len(data)
		log.Debugf("Did not receive needed header data, need %d more bytes", headerMissing)
		return proxylib.MORE, headerMissing
	}

	bodyLength := binary.BigEndian.Uint32(data[8:12])

	keyLength := binary.BigEndian.Uint16(data[2:4])
	extrasLength := data[4]

	if keyLength > 0 {
		neededData := headerSize + int(keyLength) + int(extrasLength)
		if neededData > len(data) {
			keyMissing := neededData - len(data)
			log.Debugf("Did not receive enough bytes for key, need %d more bytes", keyMissing)
			return proxylib.MORE, keyMissing
		}
	}

	opcode, key, err := p.getOpcodeAndKey(data, extrasLength, keyLength)
	if err != 0 {
		return proxylib.ERROR, int(err)
	}

	logEntry := &cilium.LogEntry_GenericL7{
		GenericL7: &cilium.L7LogEntry{
			Proto: "binarymemcached",
			Fields: map[string]string{
				"opcode": strconv.Itoa(int(opcode)),
				"key":    string(key),
			},
		},
	}

	// we don't filter reply traffic
	if reply {
		log.Debugf("reply, passing %d bytes", len(data))
		p.connection.Log(cilium.EntryType_Response, logEntry)
		p.replyCount++
		return proxylib.PASS, int(bodyLength + headerSize)
	}

	p.requestCount++

	matches := p.connection.Matches(meta.MemcacheMeta{
		Opcode: opcode,
		Keys:   [][]byte{key},
	})
	if matches {
		p.connection.Log(cilium.EntryType_Request, logEntry)
		return proxylib.PASS, int(bodyLength + headerSize)
	}

	magic := ResponseMagic | data[0]

	// This is done to ensure in-order replies
	if p.requestCount == p.replyCount+1 {
		p.injectDeniedMessage(magic)
	} else {
		p.injectQueue = append(p.injectQueue, queuedInject{magic, p.requestCount})
	}

	p.injectQueue = append(p.injectQueue, queuedInject{magic, p.requestCount})

	p.connection.Log(cilium.EntryType_Denied, logEntry)
	return proxylib.DROP, int(bodyLength + headerSize)
}