Skip to content

Commit 6ca1342

Browse files
author
Tuncay Namli
committed
♻️ refactor: Better handling of pagination parameter in onfhir client for next page, some updates on the configuration models required for new spark-on-fhir
1 parent 2c0b845 commit 6ca1342

28 files changed

Lines changed: 513 additions & 159 deletions

File tree

‎onfhir-client/src/main/scala/io/onfhir/client/OnFhirNetworkClient.scala‎

Lines changed: 39 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,8 @@ package io.onfhir.client
22

33
import akka.actor.ActorSystem
44
import akka.http.scaladsl.Http
5-
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, Uri}
5+
import akka.http.scaladsl.model.headers.{Accept, RawHeader}
6+
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, HttpResponse, MediaRange, MediaType, Uri}
67
import akka.stream.Materializer
78
import com.typesafe.config.Config
89
import io.onfhir.api.client._
@@ -11,6 +12,7 @@ import io.onfhir.client.intrcp.{BasicAuthenticationInterceptor, BearerTokenInter
1112
import io.onfhir.client.parsers.{FHIRRequestMarshaller, FHIRResponseUnmarshaller}
1213
import org.slf4j.{Logger, LoggerFactory}
1314

15+
import java.util.UUID
1416
import scala.concurrent.{ExecutionContext, Future}
1517
import scala.util.Try
1618

@@ -85,7 +87,7 @@ case class OnFhirNetworkClient(serverBaseUrl: String, interceptors: Seq[IHttpReq
8587
* @return
8688
*/
8789
override def next[T <: FHIRPaginatedBundle](bundle: T): Future[T] = {
88-
val nextPageParams = Uri.apply(bundle.getNext()).query().toMultiMap
90+
/*val nextPageParams = Uri.apply(bundle.getNext()).query().toMultiMap
8991
val previousPageParams = bundle.request.request.queryParams
9092
// Identify the pagination parameter by comparing the "next" link's parameters with the previous request's parameters
9193
val paginationParam =
@@ -104,17 +106,46 @@ case class OnFhirNetworkClient(serverBaseUrl: String, interceptors: Seq[IHttpReq
104106
case srb: FhirSearchRequestBuilder => srb.page = Some(paginationParam._1, paginationParam._2.head)
105107
case hrb: FhirHistoryRequestBuilder => hrb.page = Some(paginationParam._1, paginationParam._2.head)
106108
}
109+
}*/
110+
111+
112+
def findNextPage(paramName:String):Option[String] = {
113+
val nextPageParams = Uri.apply(bundle.getNext()).query().toMultiMap
114+
nextPageParams.get(paramName).map(_.head)
107115
}
108116

117+
val withPaginationParam =
118+
bundle.request match {
119+
case srb: FhirSearchRequestBuilder if srb.page.isDefined =>
120+
findNextPage(srb.page.get._1).foreach(pv =>
121+
srb.page = Some(srb.page.get._1, pv)
122+
)
123+
true
124+
case hrb: FhirHistoryRequestBuilder if hrb.page.isDefined =>
125+
findNextPage(hrb.page.get._1).foreach(pv =>
126+
hrb.page = Some(hrb.page.get._1, pv)
127+
)
128+
true
129+
case _ => false
130+
}
131+
132+
val httpRequest =
133+
if(withPaginationParam)
134+
FHIRRequestMarshaller
135+
.marshallRequest(bundle.request.compileRequest(), getBaseUrl())
136+
else
137+
FHIRRequestMarshaller
138+
.marshallRequest(getSearchPage(bundle.getNext()).compileRequest(), getBaseUrl())
109139

140+
/*
110141
FHIRRequestMarshaller
111142
.marshallRequest(bundle.request.compileRequest(), getBaseUrl())
112-
.flatMap(httpRequest => executeHttpRequest(httpRequest))
113-
//.marshallRequest(bundle.request.request, serverBaseUrl)
114-
/*.flatMap(httpRequest =>
115-
executeHttpRequest(httpRequest.withUri(Uri.apply(bundle.getNext())))
116-
)*/
117-
.flatMap(httpResponse => FHIRResponseUnmarshaller.unmarshallResponse(httpResponse))
143+
.flatMap(httpRequest => executeHttpRequest(httpRequest))*/
144+
httpRequest
145+
.flatMap(executeHttpRequest)
146+
.flatMap(httpResponse =>
147+
FHIRResponseUnmarshaller.unmarshallResponse(httpResponse)
148+
)
118149
.recover {
119150
case t: Throwable =>
120151
logger.error(s"Problem while executing FHIR request '${bundle.getNext()}'!", t)

‎onfhir-client/src/main/scala/io/onfhir/client/parsers/FHIRRequestMarshaller.scala‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,8 @@ object FHIRRequestMarshaller {
167167
FHIR_INTERACTIONS.HISTORY_TYPE |
168168
FHIR_INTERACTIONS.HISTORY_SYSTEM |
169169
FHIR_INTERACTIONS.SEARCH |
170-
FHIR_INTERACTIONS.CAPABILITIES =>
170+
FHIR_INTERACTIONS.CAPABILITIES |
171+
FHIR_INTERACTIONS.GET_SEARCH_PAGE =>
171172

172173
fhirRequest.httpMethod.getOrElse(HttpMethods.GET)
173174

@@ -244,6 +245,11 @@ object FHIRRequestMarshaller {
244245

245246
temp -> (if (method == HttpMethods.GET) getQuery(fhirRequest.queryParams) else None)
246247

248+
case FHIR_INTERACTIONS.GET_SEARCH_PAGE =>
249+
fhirRequest.requestUri.split('?') match {
250+
case Array(p, query) => (basePath ++ Uri.Path(p)) -> Some(Uri.Query(query))
251+
case Array(p) => basePath ++ Uri.Path(p) -> None
252+
}
247253
case opr if opr.startsWith("$") =>
248254
var temp = basePath
249255
fhirRequest.resourceType.foreach(rt => temp = temp./(rt))

‎onfhir-client/src/test/scala/io/onfhir/client/OnFhirNetworkClientTest.scala‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ import scala.concurrent.{Await, ExecutionContext, Future}
1919
import scala.language.postfixOps
2020

2121
object OnFhirNetworkClientTest extends Specification {
22-
val baseUrl = "http://localhost:8080/fhir"
22+
val baseUrl = "http://127.0.0.1:8080/fhir"
2323
val patientWithoutId: Resource = Source.fromInputStream(getClass.getResourceAsStream("/patient-without-id.json")).mkString.parseJson
2424
val obsGlucose: Resource = Source.fromInputStream(getClass.getResourceAsStream("/observation-glucose.json")).mkString.parseJson
2525
implicit val actorSystem: ActorSystem = ActorSystem("OnFhirClientTest")
@@ -158,6 +158,21 @@ object OnFhirNetworkClientTest extends Specification {
158158
itr.next().map(_.searchResults.length) must be_==(1).await
159159
}
160160

161+
"should help retrieving a searchset page from the link" in {
162+
var bundle:FHIRSearchSetBundle =
163+
Await.result(onFhirClient.getSearchPage("/Patient?gender=male&_count=2"), 5 seconds)
164+
bundle.searchResults.length mustEqual 2
165+
bundle.hasNext() mustEqual true
166+
167+
bundle = Await.result(onFhirClient.next(bundle), 5 seconds)
168+
bundle.searchResults.length mustEqual 1
169+
bundle.hasNext() mustEqual false
170+
171+
bundle = Await.result(onFhirClient.getSearchPage("http://127.0.0.1:8080/fhir/Patient?gender=male&_count=2"), 5 seconds)
172+
bundle.searchResults.length mustEqual 2
173+
bundle.hasNext() mustEqual true
174+
}
175+
161176
"should help searching resources - starting from a page" in {
162177
var bundle: FHIRSearchSetBundle = Await.result(
163178
onFhirClient

‎onfhir-common/src/main/scala/io/onfhir/api/api.scala‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -625,6 +625,8 @@ package object api {
625625
val TRANSACTION = "transaction"
626626
val BATCH = "batch"
627627
val UNKNOWN = "unknown"
628+
// Getting FHIR Search page via the given next or previous link in search-set bundle
629+
val GET_SEARCH_PAGE = "get-search-pages"
628630
//OnFhir specific interactions
629631
val BULK_UPSERT = "bulk-upsert"
630632
}

‎onfhir-common/src/main/scala/io/onfhir/api/client/BaseFhirClient.scala‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -179,6 +179,9 @@ abstract class BaseFhirClient extends IOnFhirClient {
179179
def transaction():FhirBatchTransactionRequestBuilder = {
180180
new FhirBatchTransactionRequestBuilder(this, isBatch = false)
181181
}
182+
183+
def getSearchPage(link:String):FhirGetSearchPageRequestBuilder =
184+
new FhirGetSearchPageRequestBuilder(this, link)
182185
}
183186

184187
object FhirClientUtil {

‎onfhir-common/src/main/scala/io/onfhir/api/client/FHIRBundle.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@ abstract class FHIRPaginatedBundle(bundle:Resource, val request:FhirRequestBuild
7272
* FHIR Bundle that represents a search set
7373
* @param bundle
7474
*/
75-
class FHIRSearchSetBundle(bundle:Resource, override val request:FhirSearchRequestBuilder) extends FHIRPaginatedBundle(bundle, request) {
75+
class FHIRSearchSetBundle(bundle:Resource, override val request:FHIRSearchSetReturningRequestBuilder) extends FHIRPaginatedBundle(bundle, request) {
7676

7777
/**
7878
* Search results

‎onfhir-common/src/main/scala/io/onfhir/api/client/FHIRSearchSetReturningRequestBuilder.scala‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
package io.onfhir.api.client
2+
3+
import io.onfhir.api.model.{FHIRRequest, FHIRResponse}
4+
5+
import scala.concurrent.{ExecutionContext, Future}
6+
import scala.util.{Failure, Success}
7+
8+
abstract class FHIRSearchSetReturningRequestBuilder(onFhirClient: IOnFhirClient, request: FHIRRequest)
9+
extends FhirSearchLikeRequestBuilder(onFhirClient, request)
10+
with IFhirBundleReturningRequestBuilder {
11+
12+
/**
13+
*
14+
* @param fhirResponse
15+
* @return
16+
*/
17+
override def constructBundle(fhirResponse: FHIRResponse): FHIRSearchSetBundle = {
18+
try {
19+
new FHIRSearchSetBundle(fhirResponse.responseBody.get, this)
20+
} catch {
21+
case e: Throwable =>
22+
throw FhirClientException("Invalid search result bundle!", Some(fhirResponse))
23+
}
24+
}
25+
26+
27+
/**
28+
* Send the FHIR search request and return the FHIR Bundle returned in the response parsed as FHIRSearchSetBundle if successfull, otherwise throw FhirClientException
29+
*
30+
* @param executionContext Execution context
31+
* @return
32+
*/
33+
@throws[FhirClientException]
34+
def executeAndReturnBundle()(implicit executionContext: ExecutionContext): Future[FHIRSearchSetBundle] = {
35+
execute()
36+
.map(r => {
37+
if (r.httpStatus.isFailure() || r.responseBody.isEmpty)
38+
throw FhirClientException("Problem in FHIR search!", Some(r))
39+
constructBundle(r)
40+
})
41+
}
42+
43+
/**
44+
* Returns Scala iterator where you can iterate over search results page by page
45+
*
46+
* @param executionContext Execution context
47+
* @return
48+
*/
49+
def toIterator()(implicit executionContext: ExecutionContext): Iterator[Future[FHIRSearchSetBundle]] = {
50+
new SearchSetIterator(this)
51+
}
52+
53+
/**
54+
* Send the FHIR request and paginate over the whole result set by retrieving next page until there is no further and merge them into FHIRSearchSetBundle
55+
*
56+
* @param executionContext
57+
* @return
58+
*/
59+
def executeAndMergeBundle()(implicit executionContext: ExecutionContext): Future[FHIRSearchSetBundle] = {
60+
getMergedBundle(executeAndReturnBundle())
61+
}
62+
63+
/**
64+
*
65+
* @param bundle
66+
* @param ec
67+
* @return
68+
*/
69+
private def getMergedBundle(bundle: Future[FHIRSearchSetBundle])(implicit ec: ExecutionContext): Future[FHIRSearchSetBundle] = {
70+
bundle.flatMap {
71+
case r if r.hasNext() =>
72+
getMergedBundle(onFhirClient.next(r))
73+
.map(r2 =>
74+
r2.mergeResults(r)
75+
)
76+
case r =>
77+
Future.apply(r)
78+
}
79+
}
80+
}
81+
82+
83+
/**
84+
*
85+
* @param rb
86+
* @param executionContext
87+
*/
88+
class SearchSetIterator(rb: FHIRSearchSetReturningRequestBuilder)(implicit executionContext: ExecutionContext) extends Iterator[Future[FHIRSearchSetBundle]] {
89+
var latestBundle: Option[FHIRSearchSetBundle] = None
90+
91+
override def hasNext: Boolean = latestBundle.forall(_.hasNext())
92+
93+
override def next(): Future[FHIRSearchSetBundle] = {
94+
latestBundle match {
95+
case None =>
96+
val temp = rb.executeAndReturnBundle()
97+
temp onComplete {
98+
case Success(v) => latestBundle = Some(v)
99+
case Failure(exception) =>
100+
}
101+
temp
102+
case Some(b) =>
103+
val temp = rb.onFhirClient.next(b)
104+
temp onComplete {
105+
case Success(v) => latestBundle = Some(v)
106+
case Failure(exception) =>
107+
}
108+
temp
109+
}
110+
}
111+
}

‎onfhir-common/src/main/scala/io/onfhir/api/client/FhirGetSearchPageRequestBuilder.scala‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package io.onfhir.api.client
2+
3+
import io.onfhir.api.FHIR_INTERACTIONS
4+
import io.onfhir.api.model.FHIRRequest
5+
6+
import scala.concurrent.{ExecutionContext, Future}
7+
8+
/**
9+
* For requests to load a next/previous page link in a search pagination
10+
* @param onFhirClient OnFhir client
11+
* @param link Partial/whole link for the search result set page (if partial the remaining part from FHIR server base-url should be given)
12+
* e.g. Test HAPI server example :
13+
* - whole: http://localhost:8080/fhir?_getpages=...&_getpageoffset=10
14+
* - partial: ?_getpages=...&_getpageoffset=10
15+
* e.g. With some specific path
16+
* - whole: http://localhost:8080/fhir/searchset/44225?_page=3
17+
* - partial: /searchset/44225?_page=3
18+
*/
19+
class FhirGetSearchPageRequestBuilder(onFhirClient: IOnFhirClient, link:String)
20+
extends FHIRSearchSetReturningRequestBuilder(onFhirClient,
21+
FHIRRequest(
22+
interaction = FHIR_INTERACTIONS.GET_SEARCH_PAGE,
23+
requestUri =
24+
if(link.startsWith(onFhirClient.getBaseUrl()))
25+
link.drop(onFhirClient.getBaseUrl().length)
26+
else if(link.head == '/' || link.head == '?')
27+
link
28+
else
29+
throw new IllegalArgumentException("The link for the search page should be either whole link starting with target FHIR server base url or partial link starting as path '/' or directly with query '?'")
30+
)) {
31+
32+
33+
/**
34+
*
35+
* @param parsedQuery
36+
* @return
37+
*/
38+
override def where(parsedQuery: Map[String, List[String]]): This = {
39+
throw new IllegalAccessError("Do not use this operation for this builder")
40+
}
41+
42+
override def where(param: String, value: String*): This = {
43+
throw new IllegalAccessError("Do not use this operation for this builder")
44+
}
45+
//Do nothing
46+
override protected def compile():Unit = {
47+
48+
}
49+
}
50+

‎onfhir-common/src/main/scala/io/onfhir/api/client/FhirRequestBuilder.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ object FhirRequestBuilder {
2121
.executeAndReturnResourceOption()
2222
}
2323

24-
implicit def toExecutionReturnBundle(fhirSearchRequestBuilder: FhirSearchRequestBuilder)(implicit ex:ExecutionContext):Future[FHIRSearchSetBundle] = {
24+
implicit def toExecutionReturnBundle(fhirSearchRequestBuilder: FHIRSearchSetReturningRequestBuilder)(implicit ex:ExecutionContext):Future[FHIRSearchSetBundle] = {
2525
fhirSearchRequestBuilder
2626
.executeAndReturnBundle()
2727
}

0 commit comments

Comments
 (0)