payload: {"action":"closed","issue":{"url":"https://api.github.com/repos/dapr/components-contrib/issues/154","repository_url":"https://api.github.com/repos/dapr/components-contrib","labels_url":"https://api.github.com/repos/dapr/components-contrib/issues/154/labels{/name}","comments_url":"https://api.github.com/repos/dapr/components-contrib/issues/154/comments","events_url":"https://api.github.com/repos/dapr/components-contrib/issues/154/events","html_url":"https://github.com/dapr/components-contrib/issues/154","id":536059373,"node_id":"MDU6SXNzdWU1MzYwNTkzNzM=","number":154,"title":"Upgrade EventHubs input binding to process past events","user":{"login":"bd-g","id":49082060,"node_id":"MDQ6VXNlcjQ5MDgyMDYw","avatar_url":"https://avatars2.githubusercontent.com/u/49082060?v=4","gravatar_id":"","url":"https://api.github.com/users/bd-g","html_url":"https://github.com/bd-g","followers_url":"https://api.github.com/users/bd-g/followers","following_url":"https://api.github.com/users/bd-g/following{/other_user}","gists_url":"https://api.github.com/users/bd-g/gists{/gist_id}","starred_url":"https://api.github.com/users/bd-g/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bd-g/subscriptions","organizations_url":"https://api.github.com/users/bd-g/orgs","repos_url":"https://api.github.com/users/bd-g/repos","events_url":"https://api.github.com/users/bd-g/events{/privacy}","received_events_url":"https://api.github.com/users/bd-g/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":2,"created_at":"2019-12-10T23:49:10Z","updated_at":"2019-12-11T16:48:21Z","closed_at":"2019-12-11T16:48:21Z","author_association":"NONE","body":"Currently, the event hubs input binding with Dapr only supports ingestion of events that occur after the dapr runtime is started. If an app was taken down for maintenance, and released again after, let's say, 6 hours, there is no way to tell dapr to feed the past 6 hours of events plus future events to that app. Functionality should be added if possible to accommodate this.\r\n\r\n**bindings/azure/eventhubs/eventhubs.go**\r\n```\r\n// Read gets messages from eventhubs in a non-blocking fashion\r\nfunc (a *AzureEventHubs) Read(handler func(*bindings.ReadResponse) error) error {\r\n\tcallback := func(c context.Context, event *eventhub.Event) error {\r\n\t\tif a.metadata.MessageAge != \"\" && event.SystemProperties != nil && event.SystemProperties.EnqueuedTime != nil {\r\n\t\t\tenqTime := *event.SystemProperties.EnqueuedTime\r\n\t\t\td, err := time.ParseDuration(a.metadata.MessageAge)\r\n\t\t\tif err != nil {\r\n\t\t\t\tlog.Errorf(\"error parsing duration: %s\", err)\r\n\t\t\t\treturn nil\r\n\t\t\t} else if time.Now().UTC().Sub(enqTime) > d {\r\n\t\t\t\treturn nil\r\n\t\t\t}\r\n\t\t}\r\n\t\tif event != nil {\r\n\t\t\thandler(&bindings.ReadResponse{\r\n\t\t\t\tData: event.Data,\r\n\t\t\t})\r\n\t\t}\r\n\r\n\t\treturn nil\r\n\t}\r\n\r\n\tctx := context.Background()\r\n\truntimeInfo, err := a.hub.GetRuntimeInformation(ctx)\r\n\tif err != nil {\r\n\t\treturn err\r\n\t}\r\n\r\n\tops := []eventhub.ReceiveOption{\r\n\t\t**eventhub.ReceiveWithLatestOffset(),** **here is the issue, this shouldn't be hardcoded**\r\n\t}\r\n\r\n\tif a.metadata.ConsumerGroup != \"\" {\r\n\t\tlog.Infof(\"eventhubs: using consumer group %s\", a.metadata.ConsumerGroup)\r\n\t\tops = append(ops, eventhub.ReceiveWithConsumerGroup(a.metadata.ConsumerGroup))\r\n\t}\r\n\r\n\tfor _, partitionID := range runtimeInfo.PartitionIDs {\r\n\t\t_, err := a.hub.Receive(ctx, partitionID, callback, ops...)\r\n\t\tif err != nil {\r\n\t\t\treturn err\r\n\t\t}\r\n\t}\r\n\r\n\tsignalChan := make(chan os.Signal, 1)\r\n\tsignal.Notify(signalChan, os.Interrupt, syscall.SIGTERM)\r\n\t<-signalChan\r\n\r\n\ta.hub.Close(context.Background())\r\n\treturn nil\r\n}\r\n```"}}
other: {"actor":{"display_login":"bd-g"}}