Skip to content
This repository was archived by the owner on Apr 22, 2024. It is now read-only.

[PIR-23] Increase maximum allowed data age for event replay #20

Closed
julianmclain wants to merge 3 commits into from
Closed

[PIR-23] Increase maximum allowed data age for event replay #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
66cbde8
Increase maximum allowed data age for event replay
julianmclain Jun 3, 2020
c811e32
Add http response object to Lambda
julianmclain Jun 4, 2020
ed2e950
Update error handling and test IterableLambdaEndpoint
julianmclain Jun 5, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions iterable-extension/src/main/java/com/mparticle/ext/iterable/ErrorResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
package com.mparticle.ext.iterable;

import com.fasterxml.jackson.annotation.JsonProperty;

import java.util.Map;

public class ErrorResponse {
@JsonProperty(value="statusCode", required=true)
public int statusCode;

@JsonProperty(value="body", required=true)
public Map<String, String> body;

public ErrorResponse() {}

public ErrorResponse(int statusCode, Map<String, String> body) {
this.statusCode = statusCode;
this.body = body;
}

public int getStatusCode() {
return statusCode;
}

public void setStatusCode(int statusCode) {
this.statusCode = statusCode;
}

public Map<String, String> getBody() {
return body;
}

public void setBody(Map<String, String> body) {
this.body = body;
}
}
76 changes: 28 additions & 48 deletions iterable-extension/src/main/java/com/mparticle/ext/iterable/IterableExtension.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,11 +80,7 @@ private void processPushOpens(EventProcessingRequest processingRequest) throws I
}
request.createdAt = (int) (event.getTimestamp() / 1000.0);
Response<IterableApiResponse> response = iterableService.trackPushOpen(getApiKey(processingRequest), request).execute();
if (response.isSuccessful() && !response.body().isSuccess()) {
throw new IOException(response.body().toString());
} else if (!response.isSuccessful()) {
throw new IOException("Error sending push-open to Iterable: HTTP " + response.code());
}
handleIterableResponse(response);
}
}
}
Expand Down Expand Up @@ -167,11 +163,7 @@ public void processPushSubscriptionEvent(PushSubscriptionEvent event) throws IOE
}

Response<IterableApiResponse> response = iterableService.registerToken(getApiKey(event), request).execute();
if (response.isSuccessful() && !response.body().isSuccess()) {
throw new IOException(response.body().toString());
} else if (!response.isSuccessful()) {
throw new IOException("Error sending push subscription to Iterable: " + response.body().toString());
}
handleIterableResponse(response);
}

void updateUser(EventProcessingRequest request) throws IOException {
Expand Down Expand Up @@ -203,12 +195,7 @@ void updateUser(EventProcessingRequest request) throws IOException {
//this is safe due to the filters above
updateEmailRequest.newEmail = changeEvent.getAdded().get(0).getValue();
Response<IterableApiResponse> response = iterableService.updateEmail(getApiKey(request), updateEmailRequest).execute();
if (response.isSuccessful()) {
IterableApiResponse apiResponse = response.body();
if (apiResponse != null && !apiResponse.isSuccess()) {
throw new IOException("Error while calling updateEmail() on iterable: HTTP " + apiResponse.code);
}
}
handleIterableResponse(response);
}

//convert from old to new email
Expand All @@ -218,12 +205,7 @@ void updateUser(EventProcessingRequest request) throws IOException {
updateEmailRequest.currentEmail = changeEvent.getRemoved().get(0).getValue();
updateEmailRequest.newEmail = changeEvent.getAdded().get(0).getValue();
Response<IterableApiResponse> response = iterableService.updateEmail(getApiKey(request), updateEmailRequest).execute();
if (response.isSuccessful()) {
IterableApiResponse apiResponse = response.body();
if (apiResponse != null && !apiResponse.isSuccess()) {
throw new IOException("Error while calling updateEmail() on iterable: HTTP " + apiResponse.code);
}
}
handleIterableResponse(response);
}
}

Expand All @@ -233,12 +215,7 @@ void updateUser(EventProcessingRequest request) throws IOException {
if (!isEmpty(userUpdateRequest.email) || !isEmpty(userUpdateRequest.userId)) {
userUpdateRequest.dataFields = convertAttributes(request.getUserAttributes(), shouldCoerceStrings(request));
Response<IterableApiResponse> response = iterableService.userUpdate(getApiKey(request), userUpdateRequest).execute();
if (response.isSuccessful()) {
IterableApiResponse apiResponse = response.body();
if (apiResponse != null && !apiResponse.isSuccess()) {
throw new IOException("Error while calling updateUser() on iterable: HTTP " + apiResponse.code);
}
}
handleIterableResponse(response);
}
}
}
Expand Down Expand Up @@ -280,6 +257,9 @@ private static boolean isEmpty(CharSequence chars) {
public void processProductActionEvent(ProductActionEvent event) throws IOException {
if (event.getAction().equals(ProductActionEvent.Action.PURCHASE)) {
TrackPurchaseRequest purchaseRequest = new TrackPurchaseRequest();
if (event.getId() != null) {
purchaseRequest.id = event.getId().toString();
}
purchaseRequest.createdAt = (int) (event.getTimestamp() / 1000.0);
ApiUser apiUser = new ApiUser();
addUserIdentitiesToRequest(apiUser, event.getRequest());
Expand All @@ -294,11 +274,7 @@ public void processProductActionEvent(ProductActionEvent event) throws IOExcepti
}

Response<IterableApiResponse> response = iterableService.trackPurchase(getApiKey(event), purchaseRequest).execute();
if (response.isSuccessful() && !response.body().isSuccess()) {
throw new IOException(response.body().toString());
} else if (!response.isSuccessful()) {
throw new IOException("Error sending custom event to Iterable: HTTP " + response.code());
}
handleIterableResponse(response);
}
}

Expand Down Expand Up @@ -495,6 +471,9 @@ public ModuleRegistrationResponse processRegistrationRequest(ModuleRegistrationR
Event.Type.USER_IDENTITY_CHANGE,
Event.Type.PRODUCT_ACTION);

// Specify an unlimited maximum data age (primarily for Event Replays)
eventProcessingRegistration.setMaxDataAgeHours(-1);

eventProcessingRegistration.setSupportedEventTypes(supportedEventTypes);
response.setEventProcessingRegistration(eventProcessingRegistration);
AudienceProcessingRegistration audienceRegistration = new AudienceProcessingRegistration();
Expand Down Expand Up @@ -552,11 +531,7 @@ private boolean processSubscribeEvent(CustomEvent event) throws IOException {
return false;
}
Response<IterableApiResponse> response = iterableService.updateSubscriptions(getApiKey(event), updateRequest).execute();
if (response.isSuccessful() && !response.body().isSuccess()) {
throw new IOException(response.body().toString());
} else if (!response.isSuccessful()) {
throw new IOException("Error sending update subscriptions event to Iterable: HTTP " + response.code());
}
handleIterableResponse(response);
return true;

}
Expand Down Expand Up @@ -608,16 +583,13 @@ public void processCustomEvent(CustomEvent event) throws IOException {
}

TrackRequest request = new TrackRequest(event.getName());
request.id = event.getId().toString();
Copy link

@vbabenkoru vbabenkoru Jun 3, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is getId() nullable? If so, I'd add a null check.

request.createdAt = (int) (event.getTimestamp() / 1000.0);
request.dataFields = attemptTypeConversion(event.getAttributes());
addUserIdentitiesToRequest(request, event.getRequest());

Response<IterableApiResponse> response = iterableService.track(getApiKey(event), request).execute();
if (response.isSuccessful() && !response.body().isSuccess()) {
throw new IOException(response.body().toString());
} else if (!response.isSuccessful()) {
throw new IOException("Error sending custom event to Iterable: HTTP " + response.code());
}
handleIterableResponse(response);
}

/**
Expand Down Expand Up @@ -687,11 +659,7 @@ public void processPushMessageReceiptEvent(PushMessageReceiptEvent event) throws
}
request.createdAt = (int) (event.getTimestamp() / 1000.0);
Response<IterableApiResponse> response = iterableService.trackPushOpen(getApiKey(event), request).execute();
if (response.isSuccessful() && !response.body().isSuccess()) {
throw new IOException(response.body().toString());
} else if (!response.isSuccessful()) {
throw new IOException("Error sending push-open to Iterable: HTTP " + response.code());
}
handleIterableResponse(response);
}
}
}
Expand Down Expand Up @@ -805,4 +773,16 @@ private boolean hasBundledSDK(EventProcessingRequest processingRequest) {
integrationAttributes.getOrDefault("Iterable.sdkVersion", null) != null;
}

static void handleIterableResponse(Response<IterableApiResponse> iterableResponse) throws IOException {
if (iterableResponse.isSuccessful() && !iterableResponse.body().isSuccess()) {
throw new IOException(iterableResponse.body().toString());
} else if (!iterableResponse.isSuccessful()) {
if (iterableResponse.code() == 429) {
throw new TooManyRequestsException();
} else {
throw new IOException("Error sending custom event to Iterable: HTTP " + iterableResponse.code());
}
}
}

}
21 changes: 18 additions & 3 deletions iterable-extension/src/main/java/com/mparticle/ext/iterable/IterableLambdaEndpoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,15 @@
package com.mparticle.ext.iterable;

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
import com.mparticle.sdk.model.Message;
import com.mparticle.sdk.model.MessageSerializer;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Map;


public class IterableLambdaEndpoint implements RequestStreamHandler {
Expand All @@ -19,7 +20,21 @@ public class IterableLambdaEndpoint implements RequestStreamHandler {
@Override
public void handleRequest(InputStream input, OutputStream output, Context context) throws IOException {
Message request = serializer.deserialize(input, Message.class);
Message response = processor.processMessage(request);
serializer.serialize(output, response);

try {
Message message = processor.processMessage(request);
SuccessResponse success = new SuccessResponse(200, message);
serializer.serialize(output, success);
} catch (TooManyRequestsException e) {
Map<String, String> body = new HashMap<>();
body.put("message", "Iterable rate limit exceeded");
ErrorResponse error = new ErrorResponse(429, body);
serializer.serialize(output, error);
} catch (IOException e) {
Map<String, String> body = new HashMap<>();
body.put("message", e.getMessage());
ErrorResponse error = new ErrorResponse(500, body);
Copy link

@vbabenkoru vbabenkoru Jun 5, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think if we return this, Lambda metrics will start showing those requests as successful because they don't throw exceptions. It may be better to let the exception through for monitoring unless we know we can track them downstream.

Copy link
Author

@julianmclain julianmclain Jun 8, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's a good call. I'll stop catching those exceptions unless we get another form of monitoring setup.

serializer.serialize(output, error);
}
}
}
35 changes: 35 additions & 0 deletions iterable-extension/src/main/java/com/mparticle/ext/iterable/SuccessResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
package com.mparticle.ext.iterable;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.mparticle.sdk.model.Message;

public class SuccessResponse {
@JsonProperty(value="statusCode", required=true)
public int statusCode;

@JsonProperty(value="body", required=true)
public Message body;

public SuccessResponse() {}

public SuccessResponse(int statusCode, Message body) {
this.statusCode = statusCode;
this.body = body;
}

public int getStatusCode() {
return statusCode;
}

public void setStatusCode(int statusCode) {
this.statusCode = statusCode;
}

public Message getBody() {
return body;
}

public void setBody(Message body) {
this.body = body;
}
}
10 changes: 10 additions & 0 deletions iterable-extension/src/main/java/com/mparticle/ext/iterable/TooManyRequestsException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
package com.mparticle.ext.iterable;

import java.io.IOException;

public class TooManyRequestsException extends IOException {

public TooManyRequestsException() {
super();
}
}
25 changes: 25 additions & 0 deletions iterable-extension/src/test/java/com/mparticle/ext/iterable/IterableExtensionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import com.mparticle.sdk.model.registration.ModuleRegistrationResponse;
import com.mparticle.sdk.model.registration.Setting;
import com.mparticle.sdk.model.registration.UserIdentityPermission;
import okhttp3.ResponseBody;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
Expand Down Expand Up @@ -248,6 +249,7 @@ public void testProcessCustomEvent() throws Exception {
assertEquals("123456", argument.getValue().userId);
assertEquals("some attribute value", argument.getValue().dataFields.get("some attribute key"));
assertEquals((int) (timeStamp / 1000.0), argument.getValue().createdAt + 0);
assertNotNull(argument.getValue().id);

apiResponse.code = "anything but success";

Expand Down Expand Up @@ -623,6 +625,7 @@ public void testProcessProductActionEvent() throws Exception {
assertEquals(trackPurchaseRequest.user.userId, "123456");
assertEquals(trackPurchaseRequest.items.size(), 2);
assertEquals(trackPurchaseRequest.total, new BigDecimal(101d));
assertNotNull(trackPurchaseRequest.id);
}

@Test
Expand Down Expand Up @@ -790,6 +793,28 @@ public void testUpdateSubscriptionsEvent() throws Exception {
assertNotNull("Iterable extension should have thrown an IOException", exception);
}

@Test
public void testHandleIterableSuccess() throws IOException{
IterableApiResponse iterableApiResponse = new IterableApiResponse();
iterableApiResponse.code = IterableApiResponse.SUCCESS_MESSAGE;
Response<IterableApiResponse> r = Response.success(iterableApiResponse);
IterableExtension.handleIterableResponse(r);
}

@Test(expected = TooManyRequestsException.class)
public void testHandleIterable429() throws IOException {
IterableApiResponse iterableApiResponse = new IterableApiResponse();
Response r = Response.error(429, ResponseBody.create(null, ""));
IterableExtension.handleIterableResponse(r);
}

@Test(expected = IOException.class)
public void testHandleIterableError() throws IOException {
IterableApiResponse iterableApiResponse = new IterableApiResponse();
Response r = Response.error(500, ResponseBody.create(null, ""));
IterableExtension.handleIterableResponse(r);
}

private EventProcessingRequest createEventProcessingRequest() {
EventProcessingRequest request = new EventProcessingRequest();
Account account = new Account();
Expand Down
Loading