Skip to content

Commit 337a51e

Browse files
AdphiAdphi
committed
add SearchAsync
Signed-off-by: Adphi <[email protected]>
1 parent f61ea45 commit 337a51e

7 files changed

+401
-8
lines changed

‎.gitignore

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
.idea

‎client.go

+1
Original file line numberDiff line numberDiff line change
@@ -28,5 +28,6 @@ type Client interface {
2828
PasswordModify(*PasswordModifyRequest) (*PasswordModifyResult, error)
2929

3030
Search(*SearchRequest) (*SearchResult, error)
31+
SearchAsync(searchRequest *SearchRequest, done chan struct{}) (<-chan *SearchAsyncResponse, error)
3132
SearchWithPaging(searchRequest *SearchRequest, pagingSize uint32) (*SearchResult, error)
3233
}

‎ldap_test.go

+84
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,90 @@ func TestSearch(t *testing.T) {
8989
t.Logf("TestSearch: %s -> num of entries = %d", searchRequest.Filter, len(sr.Entries))
9090
}
9191

92+
func TestSearchAsync(t *testing.T) {
93+
l, err := DialURL(ldapServer)
94+
if err != nil {
95+
t.Fatal(err)
96+
}
97+
defer l.Close()
98+
99+
searchRequest := NewSearchRequest(
100+
baseDN,
101+
ScopeWholeSubtree, DerefAlways, 0, 0, false,
102+
filter[0],
103+
attributes,
104+
nil)
105+
106+
var entries []*Entry
107+
responses, err := l.SearchAsync(searchRequest, nil)
108+
if err != nil {
109+
t.Fatal(err)
110+
}
111+
for res := range responses {
112+
if err := res.Err(); err != nil {
113+
t.Error(err)
114+
break
115+
}
116+
if res.Closed() {
117+
break
118+
}
119+
switch res.Type {
120+
case SearchAsyncResponseTypeEntry:
121+
entries = append(entries, res.Entry)
122+
case SearchAsyncResponseTypeReferral:
123+
t.Logf("Received Referral: %s", res.Referral)
124+
case SearchAsyncResponseTypeControl:
125+
t.Logf("Received Control: %s", res.Control)
126+
}
127+
}
128+
t.Logf("TestSearch: %s -> num of entries = %d", searchRequest.Filter, len(entries))
129+
}
130+
131+
func TestSearchAsyncStop(t *testing.T) {
132+
l, err := DialURL(ldapServer)
133+
if err != nil {
134+
t.Fatal(err)
135+
}
136+
defer l.Close()
137+
138+
searchRequest := NewSearchRequest(
139+
baseDN,
140+
ScopeWholeSubtree, DerefAlways, 0, 0, false,
141+
filter[0],
142+
attributes,
143+
nil)
144+
145+
var entries []*Entry
146+
done := make(chan struct{})
147+
responses, err := l.SearchAsync(searchRequest, done)
148+
if err != nil {
149+
t.Fatal(err)
150+
}
151+
for res := range responses {
152+
if err := res.Err(); err != nil {
153+
t.Error(err)
154+
break
155+
}
156+
157+
if res.Closed() {
158+
break
159+
}
160+
close(done)
161+
switch res.Type {
162+
case SearchAsyncResponseTypeEntry:
163+
entries = append(entries, res.Entry)
164+
case SearchAsyncResponseTypeReferral:
165+
t.Logf("Received Referral: %s", res.Referral)
166+
case SearchAsyncResponseTypeControl:
167+
t.Logf("Received Control: %s", res.Control)
168+
}
169+
}
170+
if len(entries) > 1 {
171+
t.Errorf("Expected 1 entry, got %d", len(entries))
172+
}
173+
t.Logf("TestSearch: %s -> num of entries = %d", searchRequest.Filter, len(entries))
174+
}
175+
92176
func TestSearchStartTLS(t *testing.T) {
93177
l, err := DialURL(ldapServer)
94178
if err != nil {

‎search.go

+115-4
Original file line numberDiff line numberDiff line change
@@ -218,6 +218,42 @@ func (s *SearchResult) PrettyPrint(indent int) {
218218
}
219219
}
220220

221+
// SearchAsyncResponseType describes the SearchAsyncResponse content type
222+
type SearchAsyncResponseType uint8
223+
224+
const (
225+
SearchAsyncResponseTypeNone SearchAsyncResponseType = iota
226+
SearchAsyncResponseTypeEntry
227+
SearchAsyncResponseTypeReferral
228+
SearchAsyncResponseTypeControl
229+
)
230+
231+
// SearchAsyncResponse holds the server's response message to an async search request
232+
type SearchAsyncResponse struct {
233+
// Type indicates the SearchAsyncResponse type
234+
Type SearchAsyncResponseType
235+
// Entry is the received entry, only set if Type is SearchAsyncResponseTypeEntry
236+
Entry *Entry
237+
// Referral is the received referral, only set if Type is SearchAsyncResponseTypeReferral
238+
Referral string
239+
// Control is the received control, only set if Type is SearchAsyncResponseTypeControl
240+
Control Control
241+
// closed indicates that the request is finished
242+
closed bool
243+
// err holds the encountered error while processing server's response, if any
244+
err error
245+
}
246+
247+
// Closed returns true if the request is finished
248+
func (r *SearchAsyncResponse) Closed() bool {
249+
return r.closed
250+
}
251+
252+
// Err returns the encountered error while processing server's response, if any
253+
func (r *SearchAsyncResponse) Err() error {
254+
return r.err
255+
}
256+
221257
// SearchRequest represents a search request to send to the server
222258
type SearchRequest struct {
223259
BaseDN string
@@ -285,10 +321,11 @@ func NewSearchRequest(
285321
// SearchWithPaging accepts a search request and desired page size in order to execute LDAP queries to fulfill the
286322
// search request. All paged LDAP query responses will be buffered and the final result will be returned atomically.
287323
// The following four cases are possible given the arguments:
288-
// - given SearchRequest missing a control of type ControlTypePaging: we will add one with the desired paging size
289-
// - given SearchRequest contains a control of type ControlTypePaging that isn't actually a ControlPaging: fail without issuing any queries
290-
// - given SearchRequest contains a control of type ControlTypePaging with pagingSize equal to the size requested: no change to the search request
291-
// - given SearchRequest contains a control of type ControlTypePaging with pagingSize not equal to the size requested: fail without issuing any queries
324+
// - given SearchRequest missing a control of type ControlTypePaging: we will add one with the desired paging size
325+
// - given SearchRequest contains a control of type ControlTypePaging that isn't actually a ControlPaging: fail without issuing any queries
326+
// - given SearchRequest contains a control of type ControlTypePaging with pagingSize equal to the size requested: no change to the search request
327+
// - given SearchRequest contains a control of type ControlTypePaging with pagingSize not equal to the size requested: fail without issuing any queries
328+
//
292329
// A requested pagingSize of 0 is interpreted as no limit by LDAP servers.
293330
func (l *Conn) SearchWithPaging(searchRequest *SearchRequest, pagingSize uint32) (*SearchResult, error) {
294331
var pagingControl *ControlPaging
@@ -402,6 +439,80 @@ func (l *Conn) Search(searchRequest *SearchRequest) (*SearchResult, error) {
402439
}
403440
}
404441

442+
// SearchAsync performs the given search request asynchronously, it takes an optional done channel to stop the request. It returns a SearchAsyncResponse channel which will be
443+
// closed when the request finished and an error, not nil if the request to the server failed
444+
func (l *Conn) SearchAsync(searchRequest *SearchRequest, done chan struct{}) (<-chan *SearchAsyncResponse, error) {
445+
if done == nil {
446+
done = make(chan struct{})
447+
}
448+
msgCtx, err := l.doRequest(searchRequest)
449+
if err != nil {
450+
return nil, err
451+
}
452+
responses := make(chan *SearchAsyncResponse)
453+
ch := make(chan *SearchAsyncResponse)
454+
rcv := func() {
455+
for {
456+
packet, err := l.readPacket(msgCtx)
457+
if err != nil {
458+
ch <- &SearchAsyncResponse{closed: true, err: err}
459+
return
460+
}
461+
462+
switch packet.Children[1].Tag {
463+
case 4:
464+
entry := &Entry{
465+
DN: packet.Children[1].Children[0].Value.(string),
466+
Attributes: unpackAttributes(packet.Children[1].Children[1].Children),
467+
}
468+
ch <- &SearchAsyncResponse{Type: SearchAsyncResponseTypeEntry, Entry: entry}
469+
case 5:
470+
err := GetLDAPError(packet)
471+
if err != nil {
472+
ch <- &SearchAsyncResponse{closed: true, err: err}
473+
return
474+
}
475+
var response SearchAsyncResponse
476+
if len(packet.Children) == 3 {
477+
for _, child := range packet.Children[2].Children {
478+
decodedChild, err := DecodeControl(child)
479+
if err != nil {
480+
responses <- &SearchAsyncResponse{closed: true, err: fmt.Errorf("failed to decode child control: %s", err)}
481+
return
482+
}
483+
response = SearchAsyncResponse{Type: SearchAsyncResponseTypeControl, Control: decodedChild}
484+
}
485+
}
486+
response.closed = true
487+
ch <- &response
488+
return
489+
case 19:
490+
ch <- &SearchAsyncResponse{Type: SearchAsyncResponseTypeReferral, Referral: packet.Children[1].Children[0].Value.(string)}
491+
}
492+
}
493+
}
494+
go func() {
495+
defer l.finishMessage(msgCtx)
496+
defer close(responses)
497+
go rcv()
498+
for {
499+
select {
500+
case <-done:
501+
responses <- &SearchAsyncResponse{
502+
closed: true,
503+
}
504+
return
505+
case res := <-ch:
506+
responses <- res
507+
if res.Closed() {
508+
return
509+
}
510+
}
511+
}
512+
}()
513+
return responses, nil
514+
}
515+
405516
// unpackAttributes will extract all given LDAP attributes and it's values
406517
// from the ber.Packet
407518
func unpackAttributes(children []*ber.Packet) []*EntryAttribute {

‎v3/client.go

+1
Original file line numberDiff line numberDiff line change
@@ -28,5 +28,6 @@ type Client interface {
2828
PasswordModify(*PasswordModifyRequest) (*PasswordModifyResult, error)
2929

3030
Search(*SearchRequest) (*SearchResult, error)
31+
SearchAsync(searchRequest *SearchRequest, done chan struct{}) (<-chan *SearchAsyncResponse, error)
3132
SearchWithPaging(searchRequest *SearchRequest, pagingSize uint32) (*SearchResult, error)
3233
}

‎v3/ldap_test.go

+84
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,90 @@ func TestSearch(t *testing.T) {
8989
t.Logf("TestSearch: %s -> num of entries = %d", searchRequest.Filter, len(sr.Entries))
9090
}
9191

92+
func TestSearchAsync(t *testing.T) {
93+
l, err := DialURL(ldapServer)
94+
if err != nil {
95+
t.Fatal(err)
96+
}
97+
defer l.Close()
98+
99+
searchRequest := NewSearchRequest(
100+
baseDN,
101+
ScopeWholeSubtree, DerefAlways, 0, 0, false,
102+
filter[0],
103+
attributes,
104+
nil)
105+
106+
var entries []*Entry
107+
responses, err := l.SearchAsync(searchRequest, nil)
108+
if err != nil {
109+
t.Fatal(err)
110+
}
111+
for res := range responses {
112+
if err := res.Err(); err != nil {
113+
t.Error(err)
114+
break
115+
}
116+
if res.Closed() {
117+
break
118+
}
119+
switch res.Type {
120+
case SearchAsyncResponseTypeEntry:
121+
entries = append(entries, res.Entry)
122+
case SearchAsyncResponseTypeReferral:
123+
t.Logf("Received Referral: %s", res.Referral)
124+
case SearchAsyncResponseTypeControl:
125+
t.Logf("Received Control: %s", res.Control)
126+
}
127+
}
128+
t.Logf("TestSearch: %s -> num of entries = %d", searchRequest.Filter, len(entries))
129+
}
130+
131+
func TestSearchAsyncStop(t *testing.T) {
132+
l, err := DialURL(ldapServer)
133+
if err != nil {
134+
t.Fatal(err)
135+
}
136+
defer l.Close()
137+
138+
searchRequest := NewSearchRequest(
139+
baseDN,
140+
ScopeWholeSubtree, DerefAlways, 0, 0, false,
141+
filter[0],
142+
attributes,
143+
nil)
144+
145+
var entries []*Entry
146+
done := make(chan struct{})
147+
responses, err := l.SearchAsync(searchRequest, done)
148+
if err != nil {
149+
t.Fatal(err)
150+
}
151+
for res := range responses {
152+
if err := res.Err(); err != nil {
153+
t.Error(err)
154+
break
155+
}
156+
157+
if res.Closed() {
158+
break
159+
}
160+
close(done)
161+
switch res.Type {
162+
case SearchAsyncResponseTypeEntry:
163+
entries = append(entries, res.Entry)
164+
case SearchAsyncResponseTypeReferral:
165+
t.Logf("Received Referral: %s", res.Referral)
166+
case SearchAsyncResponseTypeControl:
167+
t.Logf("Received Control: %s", res.Control)
168+
}
169+
}
170+
if len(entries) > 1 {
171+
t.Errorf("Expected 1 entry, got %d", len(entries))
172+
}
173+
t.Logf("TestSearch: %s -> num of entries = %d", searchRequest.Filter, len(entries))
174+
}
175+
92176
func TestSearchStartTLS(t *testing.T) {
93177
l, err := DialURL(ldapServer)
94178
if err != nil {

0 commit comments

Comments
 (0)